293 lines
13 KiB
Python
293 lines
13 KiB
Python
"""Dona V8.3: keyless, user-triggered navigation + optional HA phone hand-off.
|
||
|
||
Geocoding with Nominatim (low volume; cache + <=1 request/s); OSRM demo for
|
||
personal experimentation, switch to a private OSRM for reliable routes.
|
||
No geolocation is inferred. No credentials are stored in this module.
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import math
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import threading
|
||
import time
|
||
from urllib.parse import urlencode, urlsplit
|
||
|
||
import requests
|
||
from flask import Blueprint, jsonify, request
|
||
|
||
MAX_DEST=180
|
||
_AGENT='DonaNavigation/8.3 (private household assistant; contact: '+os.getenv('DONA_NAV_CONTACT','local-admin')+')'
|
||
_NOMINATIM='https://nominatim.openstreetmap.org'
|
||
_OSRM_DEMO='https://router.project-osrm.org'
|
||
_LOCK=threading.RLock()
|
||
_NEXT_REQUEST={'geo':0.0,'route':0.0}
|
||
_CACHE={}
|
||
_CACHE_READY=False
|
||
_CACHE_LIMIT=400
|
||
_CACHE_TTL=30*86400
|
||
|
||
class NavigationError(RuntimeError):
|
||
pass
|
||
|
||
def _config_url(name, default):
|
||
val=os.getenv(name,default).strip().rstrip('/')
|
||
u=urlsplit(val)
|
||
if not (u.scheme in ('http','https') and u.hostname and not u.username and not u.password and not u.fragment and not u.query):
|
||
raise NavigationError(f'Adresse {name} invalide.')
|
||
return val
|
||
|
||
def _cache_file():
|
||
return Path(os.getenv('DONA_NAV_CACHE_FILE','/opt/alfred/app/cache/dona_navigation.json'))
|
||
|
||
def _get_cache():
|
||
global _CACHE_READY
|
||
if not _CACHE_READY:
|
||
try:
|
||
raw=json.loads(_cache_file().read_text(encoding='utf-8'))
|
||
if isinstance(raw,dict):_CACHE.update({str(k):v for k,v in raw.items() if isinstance(v,dict)})
|
||
except (OSError,ValueError): pass
|
||
_CACHE_READY=True
|
||
return _CACHE
|
||
|
||
def _save_cache():
|
||
try:
|
||
path=_cache_file();path.parent.mkdir(parents=True,exist_ok=True)
|
||
entries=sorted(_CACHE.items(),key=lambda x:x[1].get('at',0),reverse=True)[:_CACHE_LIMIT]
|
||
tmp=path.with_name(path.name+'.tmp')
|
||
tmp.write_text(json.dumps(dict(entries),ensure_ascii=False,separators=(',',':')),encoding='utf-8')
|
||
tmp.replace(path)
|
||
except OSError:
|
||
pass # caching is optional; never block navigation
|
||
|
||
def _cached(key):
|
||
with _LOCK:
|
||
rec=_get_cache().get(key)
|
||
if rec and time.time()-float(rec.get('at',0))<_CACHE_TTL:
|
||
return rec.get('data')
|
||
return None
|
||
|
||
def _remember(key,data):
|
||
with _LOCK:
|
||
_get_cache()[key]={'at':time.time(),'data':data}
|
||
_save_cache()
|
||
|
||
def _request(kind, base, suffix, params):
|
||
# All endpoint hosts are configured by the server administrator, never by callers.
|
||
with _LOCK:
|
||
wait=max(0.0,_NEXT_REQUEST[kind]-time.monotonic())
|
||
if wait:time.sleep(wait)
|
||
_NEXT_REQUEST[kind]=time.monotonic()+1.1
|
||
try:
|
||
resp=requests.get(base+suffix,params=params,timeout=(4,11),headers={
|
||
'User-Agent':_AGENT, 'Accept':'application/json', 'Accept-Language':'fr'})
|
||
resp.raise_for_status()
|
||
if len(resp.content)>1_500_000:raise NavigationError('Réponse cartographique trop volumineuse.')
|
||
return resp.json()
|
||
except (requests.RequestException,ValueError) as exc:
|
||
raise NavigationError('Service cartographique momentanément indisponible.') from exc
|
||
|
||
def _place(value):
|
||
if not isinstance(value,str) or not 2<=len(value.strip())<=MAX_DEST:
|
||
raise NavigationError('Indique une destination de 2 à 180 caractères.')
|
||
text=' '.join(value.strip().split())
|
||
if re.search(r'[\x00-\x1f]',text):raise NavigationError('Destination invalide.')
|
||
return text
|
||
|
||
def _coord(lat,lon):
|
||
try: a,b=float(lat),float(lon)
|
||
except (ValueError,TypeError):raise NavigationError('Coordonnées invalides.') from None
|
||
if not math.isfinite(a) or not math.isfinite(b) or not (-90<=a<=90 and -180<=b<=180):
|
||
raise NavigationError('Coordonnées hors limites.')
|
||
return round(a,6),round(b,6)
|
||
|
||
def search_location(query):
|
||
query=_place(query)
|
||
key='geo:'+query.casefold()
|
||
hit=_cached(key)
|
||
if hit is not None:return hit
|
||
base=_config_url('DONA_NAV_GEOCODER_URL',_NOMINATIM)
|
||
raw=_request('geo',base,'/search',{'q':query,'format':'jsonv2','limit':3,'accept-language':'fr'})
|
||
results=[]
|
||
for entry in raw if isinstance(raw,list) else []:
|
||
if not isinstance(entry,dict):continue
|
||
try:lat,lon=_coord(entry.get('lat'),entry.get('lon'))
|
||
except NavigationError:continue
|
||
results.append({'name':str(entry.get('display_name') or query)[:220], 'lat':lat,'lon':lon})
|
||
_remember(key,results)
|
||
return results
|
||
|
||
def _origin_text(origin=None):
|
||
# No guessed location, ever. Caller or administrator must provide one.
|
||
val=(origin or os.getenv('DONA_NAV_ORIGIN','')).strip()
|
||
return _place(val) if val else None
|
||
|
||
def external_links(dest,dest_point=None):
|
||
query=_place(dest)
|
||
if dest_point:
|
||
lat,lon=_coord(dest_point['lat'],dest_point['lon'])
|
||
waze='https://www.waze.com/ul?'+urlencode({'ll':f'{lat},{lon}','navigate':'yes','utm_source':'DonaNavigation'})
|
||
maps='https://www.google.com/maps/dir/?'+urlencode({'api':'1','destination':f'{lat},{lon}','travelmode':'driving'})
|
||
else:
|
||
waze='https://www.waze.com/ul?'+urlencode({'q':query,'navigate':'yes','utm_source':'DonaNavigation'})
|
||
maps='https://www.google.com/maps/dir/?'+urlencode({'api':'1','destination':query,'travelmode':'driving'})
|
||
return {'waze':waze,'google_maps':maps}
|
||
|
||
def _route_between(start,end):
|
||
key=f"route:{start['lat']},{start['lon']}:{end['lat']},{end['lon']}"
|
||
hit=_cached(key)
|
||
if hit is not None:return hit
|
||
base=_config_url('DONA_NAV_ROUTER_URL',_OSRM_DEMO)
|
||
coords=f"{start['lon']},{start['lat']};{end['lon']},{end['lat']}"
|
||
raw=_request('route',base,'/route/v1/driving/'+coords,{'overview':'simplified','geometries':'geojson','steps':'false'})
|
||
routes=raw.get('routes',[]) if isinstance(raw,dict) and raw.get('code')=='Ok' else []
|
||
if not routes:raise NavigationError('Aucun itinéraire routier calculable.')
|
||
r=routes[0]
|
||
coords=(r.get('geometry') or {}).get('coordinates') or []
|
||
line=[]
|
||
for p in coords[:700]:
|
||
if not isinstance(p,list) or len(p)<2:continue
|
||
lat,lon=_coord(p[1],p[0]);line.append([lat,lon])
|
||
# Need a meaningful route before returning a claimed ETA.
|
||
if len(line)<2:raise NavigationError('Géométrie d’itinéraire indisponible.')
|
||
meters=float(r['distance']);seconds=float(r['duration'])
|
||
if any(not math.isfinite(x) or x<0 for x in (meters,seconds)):
|
||
raise NavigationError('Durée de trajet invalide.')
|
||
out={'distance_km':round(meters/1000,1),'duration_min':round(seconds/60),'line':line,
|
||
'traffic_included':False,'provider':'OSRM'}
|
||
_remember(key,out)
|
||
return out
|
||
|
||
def build(destination,origin=None):
|
||
dest=_place(destination)
|
||
result={'destination':dest,'origin':None,'place':None,'route':None,'note':'', 'links':external_links(dest),
|
||
'attribution':'© OpenStreetMap contributors'}
|
||
try:
|
||
hits=search_location(dest)
|
||
if hits:
|
||
result['place']=hits[0];result['links']=external_links(dest,hits[0])
|
||
else:result['note']='Adresse non trouvée ; ouvre Waze pour vérifier la destination.'
|
||
except NavigationError as exc:result['note']=str(exc)+' Le lien Waze reste disponible.'
|
||
try: start=_origin_text(origin)
|
||
except NavigationError as exc:result['note']=str(exc);start=None
|
||
if start and result['place']:
|
||
result['origin']=start
|
||
try:
|
||
matches=search_location(start)
|
||
if matches:result['route']=_route_between(matches[0],result['place']);result['origin_place']=matches[0]
|
||
else:result['note']='Point de départ introuvable : trajet non calculé.'
|
||
except NavigationError as exc:result['note']=str(exc)+' Tu peux ouvrir Waze pour calculer l’itinéraire.'
|
||
elif not start and not result['note']:
|
||
result['note']='Renseigne un point de départ pour calculer la durée. Sans départ, seule la destination est affichée.'
|
||
return result
|
||
|
||
# Explicit speech routing. Does not intercept ordinary conversation or media controls.
|
||
_NAV_RE=re.compile(r'\b(?:waze|gps|google maps|itin[eé]raire|trajet|navigation|destination|route vers|temps pour (?:aller|me rendre)|aller [aà] )\b',re.I)
|
||
_SEND_RE=re.compile(r'\b(?:envoie|envoies|transf[eè]re|partage|sur mon t[eé]l[eé]phone|à mon t[eé]l[eé]phone)\b',re.I)
|
||
|
||
def is_request(text):
|
||
return isinstance(text,str) and 3<=len(text)<500 and bool(_NAV_RE.search(text))
|
||
|
||
def is_send(text):return bool(_SEND_RE.search(text or ''))
|
||
|
||
def extract_destination(text):
|
||
s=str(text or '').strip()
|
||
s=re.sub(r'^(?:dona|alfred)[,\s]+','',s,flags=re.I)
|
||
patterns=[r'^.*?\b(?:itin[eé]raire|trajet|destination|route)\s+(?:pour|vers|jusqu[\'’]?[aà]|[aà])\s+',
|
||
r'^.*?\b(?:aller|rendre)\s+(?:[aà]|au|aux|en)\s+',
|
||
r'^.*?\b(?:waze|google maps|gps)\s+(?:pour|vers|jusqu[\'’]?[aà]|[aà])\s+',
|
||
r'^.*?\b(?:jusqu[\'’]?[aà])\s+']
|
||
for pat in patterns:
|
||
found=re.search(pat,s,re.I)
|
||
if found:s=s[found.end():];break
|
||
s=re.sub(r'\s+(?:sur|dans)\s+(?:waze|google maps)\s*$','',s,flags=re.I)
|
||
s=s.rstrip(' .?!;,')
|
||
return s[:MAX_DEST] if 2<=len(s)<=MAX_DEST else None
|
||
|
||
# DONA_NAV_FIX_V831: same validation for config display and real HA sends.
|
||
def _notify_service_name():
|
||
service=os.getenv('DONA_NAV_HA_NOTIFY_SERVICE','').strip()
|
||
match=re.fullmatch(r'notify\.([a-z0-9_]+)',service)
|
||
return match.group(1) if match else None
|
||
|
||
def _notify_url(destination,url):
|
||
service_name=_notify_service_name()
|
||
if not service_name:raise NavigationError('Configure un service Home Assistant valide dans DONA_NAV_HA_NOTIFY_SERVICE (notify.*).')
|
||
base=(os.getenv('HA_URL') or os.getenv('HOME_ASSISTANT_URL') or '').strip().rstrip('/')
|
||
token=(os.getenv('HA_TOKEN') or os.getenv('HOME_ASSISTANT_TOKEN') or '').strip()
|
||
if not base or not token:raise NavigationError('Home Assistant : HA_URL ou HA_TOKEN manquant.')
|
||
u=urlsplit(base)
|
||
if u.scheme not in ('http','https') or not u.hostname or u.username or u.password:
|
||
raise NavigationError('Adresse Home Assistant invalide.')
|
||
payload = {
|
||
'entity_id': 'notify.' + service_name,
|
||
'title': 'Itinéraire Dona',
|
||
'message': 'Ouvrir Waze pour ' + destination + ' : ' + url
|
||
}
|
||
# payload={'title':'Itinéraire Dona','message':'Ouvrir Waze pour '+destination,
|
||
# 'data':{'url':url,'clickAction':url}}
|
||
try:
|
||
res=requests.post(base+'/api/services/notify/send_message',json=payload,
|
||
#base+'/api/services/notify/'+service_name,json=payload,
|
||
headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'},timeout=10)
|
||
res.raise_for_status()
|
||
except requests.RequestException as exc:
|
||
raise NavigationError('Notification Home Assistant refusée ou inaccessible.') from exc
|
||
return True
|
||
|
||
def send_to_phone(destination):
|
||
dest=_place(destination)
|
||
# Only links generated on server; never a caller-provided URL.
|
||
return _notify_url(dest,external_links(dest)['waze'])
|
||
|
||
def _allowed_request():
|
||
if not request.is_json or request.headers.get('X-Alfred-Request')!='1':return False
|
||
origin=request.headers.get('Origin')
|
||
if origin:
|
||
try:
|
||
u=urlsplit(origin)
|
||
return u.scheme in ('http','https') and u.netloc==request.host
|
||
except ValueError:return False
|
||
return request.headers.get('Sec-Fetch-Site','same-origin') in ('same-origin','none')
|
||
|
||
def register_navigation(app,socketio):
|
||
bp=Blueprint('dona_navigation_v83',__name__)
|
||
@bp.post('/api/dona/navigation/search')
|
||
def query():
|
||
if not _allowed_request():return jsonify(ok=False,error='Requête refusée.'),403
|
||
data=request.get_json(silent=True) or {}
|
||
try:return jsonify(ok=True,**build(data.get('destination'),data.get('origin')))
|
||
except NavigationError as exc:return jsonify(ok=False,error=str(exc)),400
|
||
@bp.post('/api/dona/navigation/send')
|
||
def notify():
|
||
if not _allowed_request():return jsonify(ok=False,error='Requête refusée.'),403
|
||
try:
|
||
dest=_place((request.get_json(silent=True) or {}).get('destination'))
|
||
send_to_phone(dest)
|
||
return jsonify(ok=True,message='Notification Waze envoyée au téléphone configuré.')
|
||
except NavigationError as exc:return jsonify(ok=False,error=str(exc)),400
|
||
@bp.get('/api/dona/navigation/config')
|
||
def settings():
|
||
return jsonify(ok=True,origin=os.getenv('DONA_NAV_ORIGIN',''),
|
||
phone_configured=bool(_notify_service_name()))
|
||
app.register_blueprint(bp)
|
||
|
||
def handle_and_emit(text,socketio,state,state_lock):
|
||
dest=extract_destination(text)
|
||
if not dest:
|
||
msg='Indique une destination, par exemple : montre-moi le trajet vers Metz.'
|
||
result={'type':'dona_navigation','error':msg}
|
||
else:
|
||
result={'type':'dona_navigation', **build(dest)}
|
||
msg='Voici la destination '+dest+'.'
|
||
if result.get('route'):
|
||
msg+=f" Distance : {result['route']['distance_km']} kilomètres ; durée indicative : {result['route']['duration_min']} minutes, hors trafic."
|
||
if is_send(text):
|
||
try:send_to_phone(dest);result['sent']=True;msg+=' Le lien Waze a été envoyé au téléphone configuré.'
|
||
except NavigationError as exc:result['send_error']=str(exc);msg+=' Je n’ai pas pu envoyer la notification : '+str(exc)
|
||
with state_lock:
|
||
state['content']=result
|
||
socketio.emit('alfred_render',result)
|
||
return msg
|