From 0e8c7873d81bea931d0c30a234af620dda372656 Mon Sep 17 00:00:00 2001 From: Ian Renton Date: Sun, 2 Nov 2025 14:13:03 +0000 Subject: [PATCH] Lookup for sig_ref data #73 --- core/sig_utils.py | 83 ++++++++++++++++++++++++++++++- server/webserver.py | 38 ++++++++++++++- spotproviders/parksnpeaks.py | 6 +-- webassets/apidocs/openapi.yml | 92 +++++++++++++++++++++++++++++++++-- 4 files changed, 208 insertions(+), 11 deletions(-) diff --git a/core/sig_utils.py b/core/sig_utils.py index a915325..9eee0b0 100644 --- a/core/sig_utils.py +++ b/core/sig_utils.py @@ -1,4 +1,12 @@ -from core.constants import SIGS +import csv +from datetime import timedelta + +from pyhamtools.locator import latlong_to_locator +from requests_cache import CachedSession + +from core.constants import SIGS, HTTP_HEADERS +from core.geo_utils import wab_wai_square_to_lat_lon + # Utility function to get the icon for a named SIG. If no match is found, the "circle-question" icon will be returned. def get_icon_for_sig(sig): @@ -14,6 +22,79 @@ def get_ref_regex_for_sig(sig): return s.ref_regex return None +# Cache for SIG ref lookups +SIG_REF_DATA_CACHE_TIME_DAYS = 30 +SIG_REF_DATA_CACHE = CachedSession("cache/sig_ref_lookup_cache", + expire_after=timedelta(days=SIG_REF_DATA_CACHE_TIME_DAYS)) + +# Look up details of a SIG reference (e.g. POTA park) such as name, lat/lon, and grid. +def get_sig_ref_info(sig, sig_ref_id): + if sig.upper() == "POTA": + data = SIG_REF_DATA_CACHE.get("https://api.pota.app/park/" + sig_ref_id, headers=HTTP_HEADERS).json() + if data: + return {"name": data["name"] if "name" in data else None, + "grid": data["grid6"] if "grid6" in data else None, + "latitude": data["latitude"] if "latitude" in data else None, + "longitude": data["longitude"] if "longitude" in data else None} + elif sig.upper() == "SOTA": + data = SIG_REF_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + sig_ref_id, headers=HTTP_HEADERS).json() + if data: + return {"name": data["name"] if "name" in data else None, + "grid": data["locator"] if "locator" in data else None, + "latitude": data["latitude"] if "latitude" in data else None, + "longitude": data["longitude"] if "longitude" in data else None} + elif sig.upper() == "WWBOTA": + data = SIG_REF_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + sig_ref_id, headers=HTTP_HEADERS).json() + if data: + return {"name": data["name"] if "name" in data else None, + "grid": data["locator"] if "locator" in data else None, + "latitude": data["lat"] if "lat" in data else None, + "longitude": data["long"] if "long" in data else None} + elif sig.upper() == "GMA" or sig.upper() == "ARLHS" or sig.upper() == "ILLW" or sig.upper() == "WCA" or sig.upper() == "MOTA" or sig.upper() == "IOTA": + data = SIG_REF_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + sig_ref_id, headers=HTTP_HEADERS).json() + if data: + return {"name": data["name"] if "name" in data else None, + "grid": data["locator"] if "locator" in data else None, + "latitude": data["latitude"] if "latitude" in data else None, + "longitude": data["longitude"] if "longitude" in data else None} + elif sig.upper() == "SIOTA": + siota_csv_data = SIG_REF_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", headers=HTTP_HEADERS) + siota_dr = csv.DictReader(siota_csv_data.content.decode().splitlines()) + for row in siota_dr: + if row["SILO_CODE"] == sig_ref_id: + return {"name": row["NAME"] if "NAME" in row else None, + "grid": row["LOCATOR"] if "LOCATOR" in row else None, + "latitude": float(row["LAT"]) if "LAT" in row else None, + "longitude": float(row["LNG"]) if "LNG" in row else None} + elif sig.upper() == "WOTA": + data = SIG_REF_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", headers=HTTP_HEADERS).json() + if data: + for feature in data["features"]: + if feature["properties"]["wotaId"] == sig_ref_id: + return {"name": feature["properties"]["title"], + "grid": feature["properties"]["qthLocator"], + "latitude": feature["geometry"]["coordinates"][1], + "longitude": feature["geometry"]["coordinates"][0]} + elif sig.upper() == "ZLOTA": + data = SIG_REF_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS).json() + if data: + for asset in data: + if asset["code"] == sig_ref_id: + return {"name": asset["name"], + "grid": latlong_to_locator(asset["y"], asset["x"], 6), + "latitude": asset["y"], + "longitude": asset["x"]} + elif sig.upper() == "WAB" or sig.upper() == "WAI": + ll = wab_wai_square_to_lat_lon(sig_ref_id) + if ll: + return {"name": sig_ref_id, + "grid": latlong_to_locator(ll[0], ll[1], 6), + "latitude": ll[0], + "longitude": ll[1]} + + return None + + # Regex matching any SIG ANY_SIG_REGEX = r"(" + r"|".join(list(map(lambda p: p.name, SIGS))) + r")" diff --git a/server/webserver.py b/server/webserver.py index a05dd28..5ddf8d9 100644 --- a/server/webserver.py +++ b/server/webserver.py @@ -12,7 +12,7 @@ from core.config import MAX_SPOT_AGE, ALLOW_SPOTTING from core.constants import BANDS, ALL_MODES, MODE_TYPES, SIGS, CONTINENTS, SOFTWARE_VERSION, UNKNOWN_BAND from core.lookup_helper import lookup_helper from core.prometheus_metrics_handler import page_requests_counter, get_metrics, api_requests_counter -from core.sig_utils import get_ref_regex_for_sig +from core.sig_utils import get_ref_regex_for_sig, get_sig_ref_info from data.sig_ref import SIGRef from data.spot import Spot @@ -43,6 +43,7 @@ class WebServer: bottle.get("/api/v1/alerts")(lambda: self.serve_alerts_api()) bottle.get("/api/v1/options")(lambda: self.serve_api(self.get_options())) bottle.get("/api/v1/status")(lambda: self.serve_api(self.status_data)) + bottle.get("/api/v1/lookup/sigref")(lambda: self.serve_sig_ref_lookup_api()) bottle.post("/api/v1/spot")(lambda: self.accept_spot()) # Routes for templated pages bottle.get("/")(lambda: self.serve_template('webpage_spots')) @@ -100,6 +101,40 @@ class WebServer: response.status = 500 return json.dumps("Error - " + str(e), default=serialize_everything) + # Look up data for a SIG reference + def serve_sig_ref_lookup_api(self): + try: + # Reject if no sig or sig_ref + query = bottle.request.query + if not "sig" in query.keys() or not "sig_ref_id" in query.keys(): + response.content_type = 'application/json' + response.status = 422 + return json.dumps("Error - sig and sig_ref_id must be provided", default=serialize_everything) + + # Reject if sig_ref format incorrect for sig + sig = query.get("sig") + sig_ref_id = query.get("sig_ref_id") + if get_ref_regex_for_sig(sig) and not re.match(get_ref_regex_for_sig(sig), sig_ref_id): + response.content_type = 'application/json' + response.status = 422 + return json.dumps("Error - '" + sig_ref_id + "' does not look like a valid reference for " + sig + ".", default=serialize_everything) + + data = get_sig_ref_info(sig, sig_ref_id) + + # 404 if we don't have any data + if not data: + response.content_type = 'application/json' + response.status = 404 + return json.dumps("Error - could not find any data for this reference.", default=serialize_everything) + + # Return success + return self.serve_api(data) + except Exception as e: + logging.error(e) + response.content_type = 'application/json' + response.status = 500 + return json.dumps("Error - " + str(e), default=serialize_everything) + # Serve a JSON API endpoint def serve_api(self, data): self.last_api_access_time = datetime.now(pytz.UTC) @@ -182,7 +217,6 @@ class WebServer: return json.dumps("Error - '" + spot.dx_grid + "' does not look like a valid Maidenhead grid.", default=serialize_everything) # Reject if sig_ref format incorrect for sig - print(spot.sig_refs[0]) if spot.sig and spot.sig_refs and len(spot.sig_refs) > 0 and spot.sig_refs[0].id and get_ref_regex_for_sig(spot.sig) and not re.match(get_ref_regex_for_sig(spot.sig), spot.sig_refs[0].id): response.content_type = 'application/json' response.status = 422 diff --git a/spotproviders/parksnpeaks.py b/spotproviders/parksnpeaks.py index d2dd91b..a7d0c86 100644 --- a/spotproviders/parksnpeaks.py +++ b/spotproviders/parksnpeaks.py @@ -38,7 +38,7 @@ class ParksNPeaks(HTTPSpotProvider): # Seen PNP spots with empty frequency, and with comma-separated thousands digits mode=source_spot["actMode"].upper(), comment=source_spot["actComments"], - sig=source_spot["actClass"], + sig=source_spot["actClass"].upper(), sig_refs=[SIGRef(id=source_spot["actSiteID"])], icon=get_icon_for_sig(source_spot["actClass"]), time=datetime.strptime(source_spot["actTime"], "%Y-%m-%d %H:%M:%S").replace( @@ -54,11 +54,11 @@ class ParksNPeaks(HTTPSpotProvider): spot.de_call = m.group(1) # Log a warning for the developer if PnP gives us an unknown programme we've never seen before - if spot.sig not in ["POTA", "SOTA", "WWFF", "SiOTA", "ZLOTA", "KRMNPA"]: + if spot.sig not in ["POTA", "SOTA", "WWFF", "SIOTA", "ZLOTA", "KRMNPA"]: logging.warn("PNP spot found with sig " + spot.sig + ", developer needs to add support for this!") # SiOTA lat/lon/grid lookup - if spot.sig == "SiOTA": + if spot.sig == "SIOTA": siota_csv_data = self.SIOTA_LIST_CACHE.get(self.SIOTA_LIST_URL, headers=HTTP_HEADERS) siota_dr = csv.DictReader(siota_csv_data.content.decode().splitlines()) for row in siota_dr: diff --git a/webassets/apidocs/openapi.yml b/webassets/apidocs/openapi.yml index 691672f..d4334f3 100644 --- a/webassets/apidocs/openapi.yml +++ b/webassets/apidocs/openapi.yml @@ -17,7 +17,7 @@ paths: /spots: get: tags: - - spots + - Spots summary: Get spots description: The main API call that retrieves spots from the system. Supply this with no query parameters to retrieve all spots known to the system. Supply query parameters to filter what is retrieved. operationId: spots @@ -247,7 +247,7 @@ paths: /alerts: get: tags: - - alerts + - Alerts summary: Get alerts description: Retrieves alerts (indications of upcoming activations) from the system. Supply this with no query parameters to retrieve all alerts known to the system. Supply query parameters to filter what is retrieved. operationId: spots @@ -355,7 +355,7 @@ paths: /status: get: tags: - - general + - General summary: Get server status description: Query information about the server for use in a diagnostics display. operationId: status @@ -432,7 +432,7 @@ paths: /options: get: tags: - - general + - General summary: Get enumeration options description: Retrieves the list of options for various enumerated types, which can be found in the spots and also provided back to the API as query parameters. While these enumerated options are defined in this spec anyway, providing them in an API call allows us to define extra parameters, like the colours associated with bands, and also allows clients to set up their filters and features without having to have internal knowledge about, for example, what bands the server knows about. operationId: options @@ -488,10 +488,92 @@ paths: example: true + /lookup/sigref: + get: + tags: + - Utilities + summary: Look up SIG Ref details + description: Perform a lookup of data about a certain reference, providing the SIG and the ID of the reference. + operationId: sigref + parameters: + - name: sig + in: query + description: Special Interest Group (SIG), e.g. outdoor activity programme such as POTA + required: true + type: string + enum: + - POTA + - SOTA + - WWFF + - WWBOTA + - GMA + - HEMA + - WCA + - MOTA + - SIOTA + - ARLHS + - ILLW + - ZLOTA + - IOTA + - WOTA + - BOTA + - WAB + - WAI + example: POTA + - name: sig_ref_id + in: query + description: ID of a reference in that SIG + required: true + type: string + example: GB-0001 + responses: + '200': + description: Success + content: + application/json: + schema: + type: object + properties: + name: + type: string + description: SIG reference name + example: Null Country Park + url: + type: string + description: SIG reference URL, which the user can look up for more information + example: "https://pota.app/#/park/GB-0001" + grid: + type: string + description: Maidenhead grid locator for the reference. + example: IO91aa + latitude: + type: number + description: Latitude of the reference, in degrees. + example: 51.2345 + longitude: + type: number + description: Longitude of the reference, in degrees. + example: -1.2345 + '404': + description: Reference not found, or SIG not supported + content: + application/json: + schema: + type: string + example: "Failed" + '422': + description: Validation error + content: + application/json: + schema: + type: string + example: "Failed" + + /spot: post: tags: - - spots + - Spots summary: Add a spot description: "Supply a new spot object, which will be added to the system. Currently, this will not be reported up the chain to a cluster, POTA, SOTA etc. This may be introduced in a future version. cURL example: `curl --request POST --header \"Content-Type: application/json\" --data '{\"dx_call\":\"M0TRT\",\"time\":1760019539, \"freq\":14200000, \"comment\":\"Test spot please ignore\", \"de_call\":\"M0TRT\"}' https://spothole.app/api/v1/spot`" operationId: spot