"""Alfred dashboard V1 — safe, read-only aggregated statuses + allowed quick actions. No credentials, IP addresses, entity ids or full HA states are sent to the browser. The dashboard probes are advisory; MCP success is never inferred from a folder. """ from __future__ import annotations from datetime import datetime import json import os from pathlib import Path import shutil import threading import time from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import Request, urlopen from zoneinfo import ZoneInfo from flask import Blueprint, jsonify, request _CACHE = {'expires': 0.0, 'value': None} _LOCK = threading.Lock() def _json_get(url: str, token: str = '', timeout: float = 2.0): parsed = urlparse(url) if parsed.scheme not in ('http', 'https') or not parsed.hostname: raise ValueError('URL non HTTP(S)') headers = {'Accept': 'application/json', 'User-Agent': 'Alfred-Dashboard/1'} if token: headers['Authorization'] = 'Bearer ' + token req = Request(url, headers=headers) with urlopen(req, timeout=timeout) as response: return json.loads(response.read(15_000_000).decode('utf-8')) def _stat_local(): disk = shutil.disk_usage(Path(__file__).resolve().parents[1]) percent = round(disk.used * 100 / disk.total) if disk.total else None mem_pct = None try: m = {} for line in Path('/proc/meminfo').read_text().splitlines(): key, _, value = line.partition(':') if key in ('MemTotal', 'MemAvailable'): m[key] = int(value.strip().split()[0]) if m.get('MemTotal'): mem_pct = round((m['MemTotal'] - m.get('MemAvailable', 0)) * 100 / m['MemTotal']) except (OSError, ValueError, ZeroDivisionError): pass try: load_1m = round(os.getloadavg()[0], 2) except (OSError, AttributeError): load_1m = None return {'disk_percent': percent, 'memory_percent': mem_pct, 'load_1m': load_1m} def _safe_float(value): try: n = float(value) return round(n, 1) if abs(n) < 10000 else None except (TypeError, ValueError): return None def _ha_data(): base = os.getenv('HA_URL', '').strip().rstrip('/') token = os.getenv('HA_TOKEN', '').strip() if not (base and token): return {'available': False, 'detail': 'Home Assistant non configuré'}, {'available': False} try: states = _json_get(base + '/api/states', token, timeout=2.5) if not isinstance(states, list): raise ValueError('États Home Assistant invalides') except (ValueError, HTTPError, URLError, OSError, TimeoutError): return {'available': False, 'detail': 'Home Assistant indisponible'}, {'available': False} lights = sum(x.get('state') == 'on' for x in states if str(x.get('entity_id','')).startswith('light.')) covers = sum(x.get('state') == 'closed' for x in states if str(x.get('entity_id','')).startswith('cover.')) t = None candidates = [] for x in states: eid = str(x.get('entity_id','')) attrs = x.get('attributes') or {} if eid.startswith('climate.'): v = _safe_float(attrs.get('current_temperature')) if v is not None: candidates.append((0 if 'salon' in eid.lower() else 2, v)) elif eid.startswith('sensor.') and attrs.get('device_class') == 'temperature': v = _safe_float(x.get('state')) if v is not None: candidates.append((1 if 'salon' in eid.lower() else 3, v)) if candidates: t = sorted(candidates, key=lambda x: x[0])[0][1] condition_map = { 'sunny':'Ensoleillé', 'clear-night':'Ciel dégagé', 'partlycloudy':'Éclaircies', 'cloudy':'Nuageux', 'rainy':'Pluvieux', 'pouring':'Fortes pluies', 'snowy':'Neigeux', 'fog':'Brouillard', 'windy':'Venteux', 'lightning':'Orageux', 'hail':'Grêle', 'exceptional':'Conditions exceptionnelles', } weather = {'available': False} weather_entities = [x for x in states if str(x.get('entity_id','')).startswith('weather.')] selected_weather = (os.getenv('HA_WEATHER_ENTITY') or os.getenv('WEATHER_ENTITY') or '').strip() if selected_weather: weather_entities.sort(key=lambda item: item.get('entity_id') != selected_weather) if weather_entities: x = weather_entities[0] attrs = x.get('attributes') or {} condition = str(x.get('state') or '') weather = { 'available': True, 'temperature': _safe_float(attrs.get('temperature')), 'condition': condition_map.get(condition, 'Conditions non précisées'), 'condition_id': condition, 'unit': str(attrs.get('temperature_unit') or '°C')[:6], } return {'available': True, 'lights_on': lights, 'covers_closed': covers, 'indoor_temperature': t}, weather def _probe(url: str): if not url: return 'unconfigured' try: _json_get(url, timeout=1.8) return 'ok' except (ValueError, HTTPError, URLError, OSError, TimeoutError): return 'offline' def _services(): ollama = (os.getenv('ALFRED_OLLAMA_URL') or os.getenv('OLLAMA_HOST') or os.getenv('OLLAMA_URL') or '').strip().rstrip('/') comfy = (os.getenv('COMFYUI_URL') or os.getenv('ALFRED_COMFYUI_URL') or '').strip().rstrip('/') hermes = bool(shutil.which('hermes') or Path('/root/.local/bin/hermes').is_file() or Path('/opt/hermes/venv/bin/hermes').is_file()) return { 'ollama': _probe(ollama + '/api/tags') if ollama else 'unconfigured', 'comfyui': _probe(comfy + '/system_stats') if comfy else 'unconfigured', # Presence of a binary is not proof of a healthy agent process. 'hermes': 'installed' if hermes else 'unknown', } def dashboard_snapshot(): now = time.monotonic() with _LOCK: if _CACHE['value'] is not None and now < _CACHE['expires']: return _CACHE['value'] home, weather = _ha_data() result = {'home': home, 'weather': weather, 'system': _stat_local(), 'services': _services(), 'updated_at': datetime.now(ZoneInfo('Europe/Paris')).isoformat()} _CACHE.update(value=result, expires=time.monotonic() + 25) return result # Only these preset actions can invoke Hermes from the browser. _QUICK_ACTIONS = { 'proxmox': 'Vérifie l’état de mes serveurs Proxmox avec les outils dont tu disposes. Si tu ne peux pas y accéder, indique-le clairement, sans inventer de résultat.', 'wazuh': 'Consulte les alertes de sécurité Wazuh récentes avec tes outils. Si l’accès échoue, dis-le clairement et ne fabrique aucune alerte.', } def _trusted_local_request(): if request.headers.get('X-Alfred-Request') != '1' or not request.is_json: return False origin = request.headers.get('Origin', '') # Browsers use Origin for JSON POST; allow trusted headless local clients # only through loopback when Origin is absent. if origin: o = urlparse(origin) return o.scheme in ('http','https') and o.netloc == request.host return request.remote_addr in ('127.0.0.1', '::1') def register_dashboard(app, quick_command=None): bp = Blueprint('alfred_home_visual_v1', __name__) @bp.get('/api/alfred/dashboard') def summary(): return jsonify(dashboard_snapshot()) @bp.post('/api/alfred/dashboard/command') def action(): if not _trusted_local_request(): return jsonify(ok=False, error='Requête non autorisée.'), 403 data = request.get_json(silent=True) or {} key = data.get('action', '') if key == 'comfyui': description = data.get('description', '') if not isinstance(description, str) or not (4 <= len(description.strip()) <= 450): return jsonify(ok=False, error='Description attendue (4 à 450 caractères).'), 400 prompt = 'Génère une image avec ComfyUI correspondant à cette description : ' + description.strip() else: prompt = _QUICK_ACTIONS.get(key) if prompt is None: return jsonify(ok=False, error='Action inconnue.'), 400 if quick_command is None: return jsonify(ok=False, error='Commande interactive indisponible.'), 503 try: ok = bool(quick_command(prompt)) except Exception: return jsonify(ok=False, error='Impossible de soumettre la commande.'), 500 if not ok: # ALFRED_IDENTITY_GLASS_V1 return jsonify(ok=False, error=os.getenv('ALFRED_NAME', 'Alfred') + ' traite déjà une autre demande.'), 409 return jsonify(ok=True, queued=True), 202 app.register_blueprint(bp)