tsunami_api.py

import math
import re
import ssl
import warnings
from urllib3.exceptions import InsecureRequestWarning
from urllib3.util import create_urllib3_context
from urllib3 import PoolManager
from utils import persistent_cache, wgs84_to_tokyo
import pandas as pd

warnings.filterwarnings("ignore", category=InsecureRequestWarning)

@persistent_cache("cache/tsunami_cache.csv")
def get_tsunami_data(lat_wgs: float, lon_wgs: float, repeat=True) -> dict[str, float]:
    """
    Retrieve tsunami data from the Kochi Prefecture Tsunami API (https://bousaimap.pref.kochi.lg.jp/kochi/req/getxmlinfo.asp) for given latitude and longitude.

    :param lat_wgs: latitude as WGS84 (API uses older Tokyo Datum internally, it will be converted automatically)
    :param lon_wgs: longitude as WGS84 (API uses older Tokyo Datum internally, it will be converted automatically)
    :param repeat: Retry by slightly adjusting coordinates, in case of no data or incomplete data, sometimes a request
                   can hit a projection voxel and no data is returned for this border.
    :return: {"time_min": float, "time_max": float, "tide_min": float, "tide_max": float, "success": bool} -- Success indicates if data was found for the location. Cached indicates if the result was retrieved from cache. The cache value can be used to limit/not limit API calls if the value was cached.
    """
    lat, lon = wgs84_to_tokyo(lat_wgs, lon_wgs)
    url = f"https://bousaimap.pref.kochi.lg.jp/kochi/req/getxmlinfo.asp?mtl=411&mcx={lon}&mcy={lat}&mps={20000}"
    ctx = create_urllib3_context(ciphers=":HIGH:!DH:!aNULL")
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    http = PoolManager(ssl_context=ctx)

    response = http.request("GET", url, headers={}, body={})
    response_str = response.data.decode('shift_jis').strip().replace("\n", "").replace("\r", "")

    # --- No Data Response ---
    if int(response_str.split("<response><result><count>")[1][0]) == 0:
        if repeat:
            return get_tsunami_data(lat - 0.0001, lon + 0.001, repeat=False)
        else:
            return {"time_min": -1.0, "time_max": -1.0, "tide_min": -1.0, "tide_max": -1.0, "success": False}

    # --- Half Data Response ---
    tide_reg = re.findall(r'&#28024;&#27700;&#28145;([\d.]+)&#65374;([\d.]+)m', response_str)
    if len(tide_reg) != 1:
        if repeat:
            return get_tsunami_data(lat - 0.0001, lon + 0.001, repeat=False)
        else:
            return {"time_min": -1.0, "time_max": -1.0, "tide_min": -1.0, "tide_max": -1.0, "success": False}

    # --- Full Data Response ---
    # Tide: 到達時間40～60分 or 到達時間60分以上
    # &#21040;&#36948;&#26178;&#38291;40&#65374;60&#20998; aka. Regex: &#21040;&#36948;&#26178;&#38291;([\d&#65374;]+)&#20998 (both casees)
    # Time: 浸水深1.0～2.0m (one case)
    # &#28024;&#27700;&#28145;1.0&#65374;2.0m aka &#28024;&#27700;&#28145;([\d.]+)&#65374;([\d.]+)m
    time = re.findall(r'&#21040;&#36948;&#26178;&#38291;([\d&#65374;]+)&#20998', response_str)[0].split("&#65374;")
    if len(time) == 2: # Range time case (40-60min)
        time_min = float(time[0])
        time_max = float(time[1])
    else: # Single time case (60min+)
        time_min = float(time[0])
        time_max = math.inf
    tide_min = float(tide_reg[0][0])
    tide_max = float(tide_reg[0][1])
    return {"time_min": time_min, "time_max": time_max, "tide_min": tide_min, "tide_max": tide_max, "success": True}

def clean_tsunami_cache():
    """
    Cleans the tsunami cache by removing failed calls.
    """
    df = pd.read_csv("cache/tsunami_cache.csv")
    failed_calls = df[df["result"].str.contains('"success": false')]
    df = df.drop(failed_calls.index)
    df.to_csv("cache/tsunami_cache.csv", index=False)

def clean_tsunami_cache_zero_val():
    """
    Cleans of zero_value tsunami cache entries, because IDK where they come from...
    """
    df = pd.read_csv("cache/tsunami_cache.csv")
    zero_val_calls = df[df["result"].str.contains('"time_min": 0.0')]
    df = df.drop(zero_val_calls.index)
    df.to_csv("cache/tsunami_cache.csv", index=False)

if __name__ == "__main__":
    lat_lon_wgs = 33.555464, 133.542192
    result = get_tsunami_data(lat_lon_wgs[0], lat_lon_wgs[1])  # Nagasaki coordinates
    print(result["time_min"], result["time_max"], result["tide_min"], result["tide_max"], result["success"])
    print(f"WGS84 Coordinates: {lat_lon_wgs[0]}, {lat_lon_wgs[1]}, Tokyo Datum Coordinates: {wgs84_to_tokyo(lat_lon_wgs[0], lat_lon_wgs[1])}")
    # Link: https://bousaimap.pref.kochi.lg.jp/kochi/req/getxmlinfo.asp?mtl=411&mcx=33.552107&mcy=133.544781&mps=20000 -> Return Nothing, but API returns 0 :(
    exit(1)