#!/usr/bin/env python3 import json import os import re import socket from pathlib import Path from flask import Flask, jsonify, request ALIASES_PATH = Path(os.environ.get("WIFI_PRESENCE_ALIASES", "/home/sebas/runtime/agent-logs/wifi-presence/aliases.json")) ALERT_PREFS_PATH = Path(os.environ.get("WIFI_PRESENCE_ALERT_PREFS", "/home/sebas/runtime/agent-logs/wifi-presence/alert-prefs.json")) SUMMARY_PATH = Path(os.environ.get("WIFI_PRESENCE_SUMMARY", "/var/www/wifi-tracker/summary.json")) LOG_PATH = Path(os.environ.get("WIFI_PRESENCE_LOG", "/home/sebas/runtime/agent-logs/wifi-presence/activity.jsonl")) STATE_PATH = Path(os.environ.get("WIFI_PRESENCE_STATE", "/home/sebas/runtime/agent-logs/wifi-presence/state.json")) DOMOTICS_CONFIG_PATH = Path(os.environ.get("DOMOTICS_CONFIG", "/home/sebas/runtime/agent-logs/domotics/devices.json")) DEFAULT_DOMOTICS_CONFIG = { 'groups': [ {'id': 'all-wiz', 'name': 'Todas las luces WiZ', 'device_ids': ['wiz-1', 'wiz-2', 'wiz-3', 'wiz-4']}, ], 'devices': [ {'id': 'wiz-1', 'type': 'wiz', 'name': 'Luz WiZ', 'ip': '192.168.1.8', 'mac': 'cc:40:85:0a:cb:56', 'capabilities': ['power', 'brightness', 'color', 'temperature']}, {'id': 'wiz-2', 'type': 'wiz', 'name': 'Luz WiZ 2', 'ip': '192.168.1.9', 'mac': 'cc:40:85:0b:e1:5e', 'capabilities': ['power', 'brightness', 'color', 'temperature']}, {'id': 'wiz-3', 'type': 'wiz', 'name': 'Luz WiZ 3', 'ip': '192.168.1.76', 'mac': '98:77:d5:11:c1:ca', 'capabilities': ['power', 'brightness', 'color', 'temperature']}, {'id': 'wiz-4', 'type': 'wiz', 'name': 'Luz WiZ 4', 'ip': '192.168.1.77', 'mac': '98:77:d5:11:b2:e0', 'capabilities': ['power', 'brightness', 'color', 'temperature']}, {'id': 'tuya-plug-1', 'type': 'tuya', 'name': 'Enchufe smart - Luces balcon', 'ip': '192.168.1.48', 'mac': '38:2c:e5:4c:99:b7', 'capabilities': ['discovered_only']}, {'id': 'tuya-ir-1', 'type': 'tuya', 'name': 'Smart infrared remote', 'ip': '192.168.1.43', 'mac': 'c4:82:e1:b4:d7:ca', 'capabilities': ['discovered_only']}, ], } app = Flask(__name__) def load_json(path: Path, default): try: return json.loads(path.read_text()) except Exception: return default def write_json_atomic(path: Path, data): path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix('.tmp') tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + '\n') tmp.replace(path) def valid_mac(mac: str) -> bool: return bool(re.fullmatch(r"[0-9a-f]{2}(:[0-9a-f]{2}){5}", mac)) def load_domotics_config(): data = load_json(DOMOTICS_CONFIG_PATH, DEFAULT_DOMOTICS_CONFIG) if not DOMOTICS_CONFIG_PATH.exists(): write_json_atomic(DOMOTICS_CONFIG_PATH, data) data.setdefault('devices', []) data.setdefault('groups', []) return data def wiz_call(ip: str, payload: dict, timeout: float = 2.0): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(timeout) try: sock.sendto(json.dumps(payload).encode(), (ip, 38899)) data, _addr = sock.recvfrom(65535) return json.loads(data.decode(errors='replace')) finally: sock.close() def wiz_get_state(ip: str): return wiz_call(ip, {'method': 'getPilot', 'params': {}}) def wiz_set_state(ip: str, params: dict): return wiz_call(ip, {'method': 'setPilot', 'params': params}) def normalize_wiz_params(data: dict): params = {} if 'state' in data: params['state'] = bool(data['state']) if 'dimming' in data: params['dimming'] = max(10, min(100, int(data['dimming']))) if 'temp' in data: params['temp'] = max(2200, min(6500, int(data['temp']))) if 'r' in data and 'g' in data and 'b' in data: params['r'] = max(0, min(255, int(data['r']))) params['g'] = max(0, min(255, int(data['g']))) params['b'] = max(0, min(255, int(data['b']))) return params def apply_aliases_to_summary(summary, aliases): devices = summary.get('devices') or [] for dev in devices: mac = (dev.get('mac') or '').lower() alias = aliases.get(mac) dev['alias'] = alias dev['display_name'] = alias or dev.get('hostname') or dev.get('mac') for ev in summary.get('recent_events') or []: mac = (ev.get('mac') or '').lower() ev['alias'] = aliases.get(mac) totals = summary.setdefault('totals', {}) totals['aliased_devices'] = len([k for k, v in aliases.items() if v]) return summary @app.get('/api/summary') def summary(): return jsonify(load_json(SUMMARY_PATH, {})) @app.get('/api/aliases') def aliases_get(): return jsonify(load_json(ALIASES_PATH, {})) @app.get('/api/alert-prefs') def alert_prefs_get(): return jsonify(load_json(ALERT_PREFS_PATH, {'new_device_alerts': True, 'devices': {}})) @app.get('/api/health') def health_get(): summary = load_json(SUMMARY_PATH, {}) return jsonify({ 'ok': True, 'summary_exists': SUMMARY_PATH.exists(), 'state_exists': STATE_PATH.exists(), 'log_exists': LOG_PATH.exists(), 'summary_updated_at': summary.get('updated_at'), 'devices_now': (summary.get('totals') or {}).get('devices_now'), }) @app.post('/api/alert-prefs') def alert_prefs_post(): data = request.get_json(force=True, silent=False) or {} prefs = load_json(ALERT_PREFS_PATH, {'new_device_alerts': True, 'devices': {}}) if 'new_device_alerts' in data: prefs['new_device_alerts'] = bool(data['new_device_alerts']) mac = str(data.get('mac', '')).strip().lower() if mac: if not valid_mac(mac): return jsonify({'ok': False, 'error': 'invalid mac'}), 400 prefs.setdefault('devices', {}).setdefault(mac, {})['alerts_enabled'] = bool(data.get('alerts_enabled', True)) write_json_atomic(ALERT_PREFS_PATH, prefs) return jsonify({'ok': True, 'prefs': prefs}) @app.post('/api/aliases') def aliases_post(): data = request.get_json(force=True, silent=False) or {} mac = str(data.get('mac', '')).strip().lower() alias = str(data.get('alias', '')).strip() if not valid_mac(mac): return jsonify({'ok': False, 'error': 'invalid mac'}), 400 aliases = load_json(ALIASES_PATH, {}) if alias: aliases[mac] = alias else: aliases.pop(mac, None) write_json_atomic(ALIASES_PATH, aliases) summary = load_json(SUMMARY_PATH, {}) if summary: write_json_atomic(SUMMARY_PATH, apply_aliases_to_summary(summary, aliases)) return jsonify({'ok': True, 'mac': mac, 'alias': aliases.get(mac)}) @app.get('/api/domotics/devices') def domotics_devices_get(): config = load_domotics_config() devices = [] for dev in config.get('devices', []): item = dict(dev) if dev.get('type') == 'wiz': try: pilot = wiz_get_state(dev['ip']) item['online'] = True item['pilot'] = pilot.get('result') or {} except Exception as exc: item['online'] = False item['error'] = str(exc) else: item['online'] = None devices.append(item) return jsonify({'devices': devices, 'groups': config.get('groups', [])}) @app.post('/api/domotics/wiz/control') def domotics_wiz_control_post(): data = request.get_json(force=True, silent=False) or {} config = load_domotics_config() params = normalize_wiz_params(data) if not params: return jsonify({'ok': False, 'error': 'no supported params'}), 400 requested_ids = data.get('device_ids') or [] requested_group = data.get('group_id') selected = [] if requested_group: group = next((g for g in config.get('groups', []) if g.get('id') == requested_group), None) if not group: return jsonify({'ok': False, 'error': 'unknown group'}), 404 requested_ids = group.get('device_ids') or [] requested_set = set(requested_ids) for dev in config.get('devices', []): if dev.get('type') != 'wiz': continue if requested_set and dev.get('id') not in requested_set: continue selected.append(dev) if not selected: return jsonify({'ok': False, 'error': 'no matching wiz devices'}), 404 results = [] for dev in selected: try: res = wiz_set_state(dev['ip'], params) verify = wiz_get_state(dev['ip']) results.append({'id': dev['id'], 'name': dev['name'], 'ip': dev['ip'], 'ok': True, 'result': res.get('result'), 'pilot': verify.get('result')}) except Exception as exc: results.append({'id': dev['id'], 'name': dev['name'], 'ip': dev['ip'], 'ok': False, 'error': str(exc)}) return jsonify({'ok': all(r.get('ok') for r in results), 'params': params, 'results': results}) if __name__ == '__main__': app.run(host='127.0.0.1', port=18891)