feat: add POST /signalid/match route with scoring and caching

config.py

@@ -517,6 +517,10 @@ ALERT_WEBHOOK_TIMEOUT = _get_env_int("ALERT_WEBHOOK_TIMEOUT", 5)
 ADMIN_USERNAME = _get_env("ADMIN_USERNAME", "admin")
 ADMIN_PASSWORD = _get_env("ADMIN_PASSWORD", "admin")
 
+# Signal identification region (affects match ranking; does not filter results)
+# Valid values: GLOBAL, EU, US, UK, AU
+REGION = _get_env("REGION", "GLOBAL")
+
 
 def configure_logging() -> None:
     """Configure application logging."""

routes/signalid.py

@@ -8,10 +8,12 @@ import urllib.parse
 import urllib.request
 from typing import Any
 
+import config
 from flask import Blueprint, Response, jsonify, request
 
 from utils.logging import get_logger
 from utils.responses import api_error
+from utils.signal_db import match_signals
 
 logger = get_logger('intercept.signalid')
 
@@ -351,3 +353,95 @@ def sigidwiki_lookup() -> Response:
         **response_payload,
     })
 
+
+_match_cache: dict[str, dict[str, Any]] = {}
+
+
+@signalid_bp.route('/match', methods=['POST'])
+def signalid_match() -> Response:
+    """Match a signal by frequency, bandwidth, and modulation against the local database."""
+    payload = request.get_json(silent=True) or {}
+
+    freq_raw = payload.get('frequency_mhz')
+    if freq_raw is None:
+        return api_error('frequency_mhz is required', 400)
+    try:
+        frequency_mhz = float(freq_raw)
+    except (TypeError, ValueError):
+        return api_error('Invalid frequency_mhz', 400)
+    if frequency_mhz <= 0:
+        return api_error('frequency_mhz must be positive', 400)
+
+    bw_raw = payload.get('bandwidth_hz')
+    bandwidth_hz: int | None = None
+    if bw_raw is not None:
+        try:
+            bandwidth_hz = int(float(bw_raw))
+        except (TypeError, ValueError):
+            return api_error('Invalid bandwidth_hz', 400)
+        if bandwidth_hz <= 0:
+            return api_error('bandwidth_hz must be positive', 400)
+
+    modulation = str(payload.get('modulation') or '').strip().upper()[:16] or None
+
+    limit_raw = payload.get('limit', 8)
+    try:
+        limit = max(1, min(int(limit_raw), 20))
+    except (TypeError, ValueError):
+        limit = 8
+
+    region = getattr(config, 'REGION', 'GLOBAL')
+
+    cache_key = f'{round(frequency_mhz, 6)}|{bandwidth_hz}|{modulation}|{limit}|{region}'
+    cached = _cache_get_match(cache_key)
+    if cached is not None:
+        return jsonify({
+            'status': 'ok',
+            'frequency_mhz': round(frequency_mhz, 6),
+            'bandwidth_hz': bandwidth_hz,
+            'modulation': modulation,
+            'cached': True,
+            **cached,
+        })
+
+    try:
+        matches = match_signals(
+            frequency_mhz=frequency_mhz,
+            bandwidth_hz=bandwidth_hz,
+            modulation=modulation,
+            region=region,
+            limit=limit,
+        )
+    except Exception as exc:
+        logger.error('Signal match failed: %s', exc)
+        return api_error('Signal match failed', 502)
+
+    response_data = {
+        'matches': matches,
+        'match_count': len(matches),
+    }
+    _cache_set_match(cache_key, response_data)
+
+    return jsonify({
+        'status': 'ok',
+        'frequency_mhz': round(frequency_mhz, 6),
+        'bandwidth_hz': bandwidth_hz,
+        'modulation': modulation,
+        'cached': False,
+        **response_data,
+    })
+
+
+def _cache_get_match(key: str) -> Any | None:
+    entry = _match_cache.get(key)
+    if not entry:
+        return None
+    if time.time() >= entry['expires']:
+        _match_cache.pop(key, None)
+        return None
+    return entry['data']
+
+
+def _cache_set_match(key: str, data: Any, ttl: int = 60) -> None:
+    _match_cache[key] = {'data': data, 'expires': time.time() + ttl}
+

tests/test_signalid_match_route.py

@@ -0,0 +1,97 @@
+"""Integration tests for POST /signalid/match route."""
+
+from __future__ import annotations
+
+import json
+import pytest
+
+
+@pytest.fixture
+def client():
+    """Minimal Flask test client with only the signalid blueprint registered."""
+    from flask import Flask
+    from routes.signalid import signalid_bp
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(signalid_bp)
+
+    with app.test_client() as c:
+        yield c
+
+
+class TestSignalidMatchRoute:
+    def test_missing_frequency_returns_400(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_invalid_frequency_returns_400(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": -1}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_valid_request_returns_200(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": 98.5}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data["status"] == "ok"
+        assert isinstance(data["matches"], list)
+        assert isinstance(data["match_count"], int)
+
+    def test_fm_broadcast_in_results_at_98mhz(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": 98.5, "modulation": "WFM"}),
+            content_type="application/json",
+        )
+        data = resp.get_json()
+        names = [m["name"] for m in data["matches"]]
+        assert "FM Broadcast Radio" in names
+
+    def test_no_matches_returns_empty_list(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": 5000.0}),
+            content_type="application/json",
+        )
+        data = resp.get_json()
+        assert data["status"] == "ok"
+        assert data["matches"] == []
+        assert data["match_count"] == 0
+
+    def test_limit_param_respected(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": 98.5, "limit": 2}),
+            content_type="application/json",
+        )
+        data = resp.get_json()
+        assert len(data["matches"]) <= 2
+
+    def test_bandwidth_hz_accepted(self, client):
+        resp = client.post(
+            "/signalid/match",
+            data=json.dumps({"frequency_mhz": 98.5, "bandwidth_hz": 200000}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 200
+
+    def test_cached_true_on_second_identical_request(self, client):
+        import routes.signalid as signalid_module
+        signalid_module._match_cache.clear()
+        payload = json.dumps({"frequency_mhz": 98.5})
+        client.post("/signalid/match", data=payload, content_type="application/json")
+        resp2 = client.post("/signalid/match", data=payload, content_type="application/json")
+        data = resp2.get_json()
+        assert data["cached"] is True