diff --git a/config/address_book.yaml b/config/address_book.yaml new file mode 100644 index 0000000..24bc81c --- /dev/null +++ b/config/address_book.yaml @@ -0,0 +1,18 @@ +# RECON Address Book — saved locations for navigation shortcuts. +# Entries are matched by name and aliases (case-insensitive). +# Add new entries by appending to the list below. + +entries: + - id: home + name: Home + aliases: + - home + - matt's house + - 214 north st + - 214 north street + address: "214 North St, Filer, ID 83328" + lat: 42.5735833 + lon: -114.6066389 + tags: + - residence + - primary diff --git a/lib/address_book.py b/lib/address_book.py new file mode 100644 index 0000000..a9cfc40 --- /dev/null +++ b/lib/address_book.py @@ -0,0 +1,132 @@ +""" +RECON Address Book — YAML-backed saved-location lookup. + +Provides named locations (home, work, etc.) that short-circuit Photon +geocoding when an exact alias match is found. + +Config: /opt/recon/config/address_book.yaml +""" + +import os +import threading + +import yaml + +from .utils import setup_logging + +logger = setup_logging('recon.address_book') + +_CONFIG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + 'config', 'address_book.yaml', +) + +_lock = threading.Lock() +_entries: list[dict] = [] +_mtime: float = 0.0 + + +def _reload_if_changed(): + """Reload the YAML file if its mtime has changed.""" + global _entries, _mtime + try: + st = os.stat(_CONFIG_PATH) + except FileNotFoundError: + logger.warning("Address book not found: %s", _CONFIG_PATH) + _entries = [] + _mtime = 0.0 + return + + if st.st_mtime == _mtime: + return + + with _lock: + # Double-check after acquiring lock + try: + st = os.stat(_CONFIG_PATH) + except FileNotFoundError: + _entries = [] + _mtime = 0.0 + return + if st.st_mtime == _mtime: + return + + with open(_CONFIG_PATH, 'r') as f: + data = yaml.safe_load(f) or {} + + raw = data.get('entries', []) + loaded = [] + for entry in raw: + # Normalise aliases to lowercase for matching + aliases = [a.lower() for a in entry.get('aliases', [])] + loaded.append({ + 'id': entry.get('id', ''), + 'name': entry.get('name', ''), + 'aliases': aliases, + 'address': entry.get('address', ''), + 'lat': entry.get('lat'), + 'lon': entry.get('lon'), + 'tags': entry.get('tags', []), + }) + _entries = loaded + _mtime = st.st_mtime + logger.info("Address book loaded: %d entries from %s", len(_entries), _CONFIG_PATH) + + +def load(): + """Ensure the address book is loaded (and refreshed if the file changed).""" + _reload_if_changed() + return _entries + + +def lookup(query: str): + """ + Look up a query against name and aliases. + + Returns dict with the matching entry plus a 'confidence' field: + - "exact": full name or alias match + - "partial": query is a substring of an alias or name (or vice versa) + - None if no match + """ + _reload_if_changed() + q = query.strip().lower() + if not q: + return None + + best = None + best_confidence = None + + for entry in _entries: + # Exact match on name + if q == entry['name'].lower(): + return {**entry, 'confidence': 'exact'} + + # Exact match on any alias + if q in entry['aliases']: + return {**entry, 'confidence': 'exact'} + + # Partial: query is substring of name/alias, or name/alias is substring of query + name_lower = entry['name'].lower() + if q in name_lower or name_lower in q: + if best is None: + best = entry + best_confidence = 'partial' + continue + + for alias in entry['aliases']: + if q in alias or alias in q: + if best is None: + best = entry + best_confidence = 'partial' + break + + if best is not None: + return {**best, 'confidence': best_confidence} + + return None + + +def list_all(): + """Return all address book entries.""" + _reload_if_changed() + return list(_entries) diff --git a/lib/address_book_api.py b/lib/address_book_api.py new file mode 100644 index 0000000..020828b --- /dev/null +++ b/lib/address_book_api.py @@ -0,0 +1,31 @@ +""" +RECON Address Book API — Flask Blueprint. + +GET /api/address_book/lookup?q= — best match or 404 +GET /api/address_book/list — all entries +""" + +from flask import Blueprint, request, jsonify + +from . import address_book + +address_book_bp = Blueprint('address_book', __name__) + + +@address_book_bp.route('/api/address_book/lookup') +def api_address_book_lookup(): + q = request.args.get('q', '').strip() + if not q: + return jsonify({'error': 'Missing q parameter'}), 400 + + result = address_book.lookup(q) + if result is None: + return '', 404 + + return jsonify(result) + + +@address_book_bp.route('/api/address_book/list') +def api_address_book_list(): + entries = address_book.list_all() + return jsonify(entries) diff --git a/lib/address_book_test.py b/lib/address_book_test.py new file mode 100644 index 0000000..e7fa7ef --- /dev/null +++ b/lib/address_book_test.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Tests for RECON address book module.""" +import sys +import os + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from lib import address_book + +TESTS = [ + ("lookup('home') → exact", + lambda: address_book.lookup("home"), + lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), + + ("lookup('Home') → exact (case-insensitive)", + lambda: address_book.lookup("Home"), + lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), + + ("lookup('214 north st') → exact via alias", + lambda: address_book.lookup("214 north st"), + lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), + + ("lookup('214 North Street') → exact via alias", + lambda: address_book.lookup("214 North Street"), + lambda r: r is not None and r['confidence'] == 'exact' and r['id'] == 'home'), + + ("lookup('nonexistent place') → None", + lambda: address_book.lookup("nonexistent place"), + lambda r: r is None), + + ("list_all() → 1 entry", + lambda: address_book.list_all(), + lambda r: isinstance(r, list) and len(r) == 1 and r[0]['id'] == 'home'), +] + +passed = 0 +failed = 0 +for name, fn, check in TESTS: + try: + result = fn() + ok = check(result) + except Exception as e: + ok = False + result = f"EXCEPTION: {e}" + + status = "PASS" if ok else "FAIL" + if ok: + passed += 1 + else: + failed += 1 + print(f" [{status}] {name}") + if not ok: + print(f" got: {result}") + +print(f"\n{passed} passed, {failed} failed") +sys.exit(0 if failed == 0 else 1) diff --git a/lib/api.py b/lib/api.py index a739ec0..49e7005 100644 --- a/lib/api.py +++ b/lib/api.py @@ -44,6 +44,11 @@ app = Flask(__name__, app.config['MAX_CONTENT_LENGTH'] = None # ZIM files can be multi-GB +# ── Address Book Blueprint ── +from .address_book_api import address_book_bp +app.register_blueprint(address_book_bp) + + # ── Navigation Constants ── KNOWLEDGE_SUBNAV = [ diff --git a/lib/nav_tools.py b/lib/nav_tools.py index f6db5e6..832ca2d 100644 --- a/lib/nav_tools.py +++ b/lib/nav_tools.py @@ -3,6 +3,10 @@ import re import requests +from .utils import setup_logging + +logger = setup_logging('recon.nav_tools') + PHOTON_URL = "http://localhost:2322" VALHALLA_URL = "http://localhost:8002" @@ -20,11 +24,26 @@ def _parse_coords(text: str): def _geocode(query: str): - """Geocode a place name via Photon. Returns (lat, lon, display_name) or raises.""" + """Geocode a place name via address book then Photon. Returns (lat, lon, display_name) or raises.""" coords = _parse_coords(query) if coords: return coords[0], coords[1], query + # ── Address book lookup (before Photon) ── + try: + from . import address_book + match = address_book.lookup(query) + if match and match['confidence'] == 'exact' and match.get('lat') and match.get('lon'): + logger.info("Address book exact match: %r → %s (%s, %s)", + query, match['name'], match['lat'], match['lon']) + return match['lat'], match['lon'], match.get('address') or match['name'] + elif match and match['confidence'] == 'partial': + logger.info("Address book partial match: %r → %s (falling through to Photon)", + query, match['name']) + except Exception as e: + logger.debug("Address book lookup failed: %s", e) + + # ── Photon geocoding ── try: resp = requests.get( f"{PHOTON_URL}/api",