228 lines
11 KiB
Python
228 lines
11 KiB
Python
"""Read-only weather adapter for Alfred: use existing HA_URL / HA_TOKEN.
|
||
|
||
Home Assistant stores current conditions in weather entities. Forecasts must be
|
||
retrieved separately with the weather.get_forecasts action (return_response).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from datetime import datetime, timedelta, time
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from . import ha_views as ha
|
||
from . import alfred_intents as intents
|
||
|
||
TZ = ZoneInfo(os.getenv('ALFRED_TIMEZONE', 'Europe/Paris'))
|
||
|
||
CONDITIONS = {
|
||
'clear-night': 'ciel dégagé', 'cloudy': 'ciel couvert',
|
||
'exceptional': 'conditions exceptionnelles', 'fog': 'brouillard',
|
||
'hail': 'grêle', 'lightning': 'orages', 'lightning-rainy': 'orages et pluie',
|
||
'partlycloudy': 'éclaircies', 'pouring': 'fortes pluies',
|
||
'rainy': 'pluie', 'snowy': 'neige', 'snowy-rainy': 'pluie et neige',
|
||
'sunny': 'ensoleillé', 'windy': 'vent', 'windy-variant': 'vent et nuages',
|
||
}
|
||
RAIN = {'rainy', 'pouring', 'lightning-rainy', 'snowy-rainy'}
|
||
|
||
|
||
def is_weather_request(text: str, previous_view: str | None = None) -> bool:
|
||
t = intents.simplify(str(text or ''))
|
||
if re.search(r'\b(meteo|previsions?|previsions? meteorologiques?|previsions? du temps|pluie|pleuvoir|pleuvra|pleut|averse|averses|orage|orages|enneigement|neige|neiger|soleil|ensoleille|ensoleillee|parapluie|quel temps|quelle meteo|beau temps|mauvais temps)\b', t):
|
||
return True
|
||
if re.search(r'\b(temperature|temperatures|combien de degres)\b.{0,24}\b(dehors|exterieur|exterieure|exterieures)\b', t):
|
||
return True
|
||
if re.search(r'\b(fait.il|fera.t.il|il fera|il fait)\b.{0,15}\b(beau|froid|chaud|dehors)\b', t):
|
||
return True
|
||
if re.search(r'\btemperature\b', t) and re.search(r'\b(demain|week.end|semaine|exterieure?|dehors|aujourd.hui)\b', t) and not re.search(r'\b(salon|chambre|cuisine|maison|bureau|salle de bain)\b',t):
|
||
return True
|
||
# Follow-ups are interpreted in the active view, not with old Hermes prompts.
|
||
if previous_view == 'weather' and len(t) < 42 and re.search(r'\b(demain|aujourd.hui|matin|apres.midi|soir|week.end|semaine|jours?|maintenant|actuellement)\b', t):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _selected_weather_entity():
|
||
entries = [s for s in ha.states() if str(s.get('entity_id', '')).startswith('weather.')]
|
||
configured = os.getenv('HA_WEATHER_ENTITY', '').strip()
|
||
if configured:
|
||
match = next((e for e in entries if e['entity_id'] == configured), None)
|
||
if not match:
|
||
raise ha.HAError(f"Entité météo {configured} introuvable dans Home Assistant.")
|
||
return match
|
||
if not entries:
|
||
raise ha.HAError('Aucune entité weather.* trouvée dans Home Assistant. Ajouter une intégration météo.')
|
||
# Prefer a conventional home weather entity, otherwise use the first available.
|
||
return next((e for e in entries if e.get('entity_id') in ('weather.home', 'weather.forecast_home')), entries[0])
|
||
|
||
|
||
def _safe_number(value):
|
||
try:
|
||
if value is None or isinstance(value, bool):
|
||
return None
|
||
return round(float(value), 1)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def _forecast(entity_id: str, typ: str):
|
||
try:
|
||
raw = ha._request('POST', '/api/services/weather/get_forecasts?return_response',
|
||
payload={'entity_id': entity_id, 'type': typ}, timeout=16)
|
||
response = raw.get('service_response', raw) if isinstance(raw, dict) else {}
|
||
ent = response.get(entity_id, {}) if isinstance(response, dict) else {}
|
||
items = ent.get('forecast', []) if isinstance(ent, dict) else []
|
||
if not isinstance(items, list):
|
||
return [], 'Format de prévision météo inattendu.'
|
||
normalized = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
raw_dt = item.get('datetime')
|
||
if not raw_dt:
|
||
continue
|
||
try:
|
||
dt = datetime.fromisoformat(str(raw_dt).replace('Z', '+00:00'))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=TZ)
|
||
dt = dt.astimezone(TZ)
|
||
except ValueError:
|
||
continue
|
||
normalized.append({
|
||
'datetime': dt.isoformat(), 'date': dt.date().isoformat(),
|
||
'condition': str(item.get('condition') or 'unknown'),
|
||
'description': CONDITIONS.get(str(item.get('condition')), 'conditions non précisées'),
|
||
'temperature': _safe_number(item.get('temperature')),
|
||
'templow': _safe_number(item.get('templow')),
|
||
'precipitation_probability': _safe_number(item.get('precipitation_probability')),
|
||
'precipitation': _safe_number(item.get('precipitation')),
|
||
'wind_speed': _safe_number(item.get('wind_speed')),
|
||
})
|
||
return normalized, None
|
||
except ha.HAError as exc:
|
||
# Some providers have daily only / hourly only forecasts.
|
||
return [], f'Prévisions {typ} indisponibles : {exc}'
|
||
|
||
|
||
def get_weather() -> dict:
|
||
item = _selected_weather_entity()
|
||
attrs = item.get('attributes') or {}
|
||
eid = item['entity_id']
|
||
current = {
|
||
'condition': str(item.get('state') or 'unknown'),
|
||
'description': CONDITIONS.get(str(item.get('state')), 'conditions non précisées'),
|
||
'temperature': _safe_number(attrs.get('temperature')),
|
||
'apparent_temperature': _safe_number(attrs.get('apparent_temperature')),
|
||
'humidity': _safe_number(attrs.get('humidity')),
|
||
'wind_speed': _safe_number(attrs.get('wind_speed')),
|
||
'pressure': _safe_number(attrs.get('pressure')),
|
||
}
|
||
daily, err_daily = _forecast(eid, 'daily')
|
||
hourly, err_hourly = _forecast(eid, 'hourly')
|
||
return {
|
||
'type': 'weather', 'entity_id': eid,
|
||
'name': str(attrs.get('friendly_name') or eid),
|
||
'temperature_unit': str(attrs.get('temperature_unit') or '°C'),
|
||
'wind_speed_unit': str(attrs.get('wind_speed_unit') or 'km/h'),
|
||
'current': current, 'daily': daily, 'hourly': hourly,
|
||
'errors': [e for e in (err_daily, err_hourly) if e],
|
||
'source': 'home_assistant', 'updated_at': datetime.now(TZ).isoformat(),
|
||
}
|
||
|
||
|
||
def _target(text: str, now: datetime) -> tuple[str, datetime, datetime, str]:
|
||
t = intents.simplify(text)
|
||
start = datetime.combine(now.date(), time.min, tzinfo=TZ)
|
||
label = 'actuellement'
|
||
days = re.search(r'\b(?:dans|sur|pour)\s+(\d{1,2})\s+jours?\b', t)
|
||
if days and 'dans ' in days.group(0):
|
||
d = min(int(days.group(1)), 14)
|
||
start += timedelta(days=d)
|
||
label = f'dans {d} jours'
|
||
elif 'week-end' in text.lower() or 'weekend' in t or 'week.end' in t:
|
||
offset = (5 - now.weekday()) % 7
|
||
start += timedelta(days=offset)
|
||
label = 'ce week-end'
|
||
return label, start, start + timedelta(days=2), 'weekend'
|
||
else:
|
||
day = intents.parse_day(text, now)
|
||
if day is not None:
|
||
start = datetime.combine(day, time.min, tzinfo=TZ)
|
||
label = 'demain' if day == now.date()+timedelta(days=1) else ('aujourd’hui' if day == now.date() else day.strftime('le %d/%m'))
|
||
kind = 'day'
|
||
if 'ce matin' in t or 'demain matin' in t:
|
||
kind = 'hours';start = start.replace(hour=6);end = start.replace(hour=12)
|
||
elif 'apres-midi' in t:
|
||
kind = 'hours';start = start.replace(hour=12);end = start.replace(hour=18)
|
||
elif 'soir' in t:
|
||
kind = 'hours';start = start.replace(hour=18);end = start + timedelta(hours=6)
|
||
else:
|
||
end = start + timedelta(days=1)
|
||
return label, start, end, kind
|
||
|
||
|
||
def _rounded_temp(n, unit='°C'):
|
||
return f'{n:g} {unit}' if n is not None else 'température indisponible'
|
||
|
||
|
||
def describe(data: dict, text: str = '') -> str:
|
||
"""Concise voice response based on actual forecast data, never invented."""
|
||
now = datetime.now(TZ)
|
||
t = intents.simplify(text)
|
||
unit = data['temperature_unit']
|
||
label, start, end, kind = _target(text, now)
|
||
weather_question = bool(re.search(r'\b(pleut|pleuvra|pleuvoir|pluie|averses|parapluie)\b', t))
|
||
days_question = bool(re.search(r'\b(semaine|7 jours|sept jours)\b', t))
|
||
now_question = (start.date() == now.date() and kind == 'day'
|
||
and not re.search(r'\b(demain|jour|mois|week|semaine)\b', t))
|
||
is_now_explicit = bool(re.search(r'\b(maintenant|en ce moment|actuellement)\b',t))
|
||
if is_now_explicit and weather_question:
|
||
c=data['current']
|
||
raining = c['condition'] in RAIN
|
||
return ('Oui, les conditions actuelles indiquent de la pluie.' if raining
|
||
else f"Actuellement, {c['description']}.")
|
||
if now_question and not weather_question and not days_question:
|
||
c=data['current']
|
||
return f"Actuellement, {c['description']}, {_rounded_temp(c['temperature'],unit)}."
|
||
if days_question:
|
||
if data['daily']:
|
||
return f"Voici les prévisions météorologiques des prochains jours, selon {data['name']}."
|
||
return 'Les prévisions journalières ne sont pas disponibles dans Home Assistant.'
|
||
candidates = [h for h in data['hourly'] if start <= datetime.fromisoformat(h['datetime']) < end]
|
||
forecasts = (candidates if kind == 'hours' and candidates else
|
||
[d for d in data['daily'] if start.date() <= datetime.fromisoformat(d['datetime']).date()
|
||
< (end.date() if end.time() == time.min else end.date() + timedelta(days=1))])
|
||
# A missing hourly forecast must not masquer itself as a reliable part-of-day prediction.
|
||
approximate = kind == 'hours' and not candidates and bool(forecasts)
|
||
if not forecasts:
|
||
if label == 'actuellement':
|
||
return f"Actuellement, {data['current']['description']}, {_rounded_temp(data['current']['temperature'], unit)}."
|
||
return f"Je n'ai pas de prévision disponible pour {label}."
|
||
if weather_question:
|
||
prob = [f['precipitation_probability'] for f in forecasts if f['precipitation_probability'] is not None]
|
||
if prob:
|
||
peak = max(prob)
|
||
response = f"Pour {label}, la probabilité de précipitations atteint {peak:g} %."
|
||
elif any(f['condition'] in RAIN for f in forecasts):
|
||
response = f"La météo prévoit de la pluie pour {label}."
|
||
else:
|
||
response = f"Aucune pluie n'est indiquée par les prévisions disponibles pour {label}, mais le risque exact n'est pas connu."
|
||
else:
|
||
first = forecasts[0]
|
||
temps = [f['temperature'] for f in forecasts if f['temperature'] is not None]
|
||
lows = [f['templow'] for f in forecasts if f['templow'] is not None]
|
||
descriptor = first['description']
|
||
if len(forecasts) > 1 and len({f['condition'] for f in forecasts}) > 1:
|
||
descriptor = 'des conditions variables'
|
||
if kind == 'hours' and candidates and temps:
|
||
response = f"Pour {label}, {descriptor}, entre {_rounded_temp(min(temps),unit)} et {_rounded_temp(max(temps),unit)}."
|
||
elif temps:
|
||
response = f"Pour {label}, {descriptor}, jusqu'à {_rounded_temp(max(temps),unit)}."
|
||
if lows:
|
||
response += f" Minimum {_rounded_temp(min(lows),unit)}."
|
||
else:
|
||
response = f"Pour {label}, {descriptor}, mais la température n'est pas disponible."
|
||
if approximate:
|
||
response += ' Cette indication est journalière, faute de prévision horaire.'
|
||
return response
|