#!/usr/bin/env python3 """ Serveur de simulation KC868-A2 — port 8080. Sert les fichiers statiques depuis ../data/ et implémente tous les endpoints /api/* avec un état en mémoire qui varie dans le temps. Lancement : python3 sim.py Accès : http://localhost:8080 """ import http.server import json import math import os import socketserver import threading import time from urllib.parse import urlparse, parse_qs # Chemin vers les fichiers web du projet DATA_DIR = os.path.join(os.path.dirname(__file__), '..', 'data') PORT = int(os.environ.get('SIM_PORT', 8080)) MIME = { '.html': 'text/html; charset=utf-8', '.js': 'application/javascript', '.css': 'text/css', '.json': 'application/json', '.ico': 'image/x-icon', } # --------------------------------------------------------------------------- # État simulé — modifiable via les endpoints POST # --------------------------------------------------------------------------- _lock = threading.Lock() state = { 'pv': 18.72, 'pvCurrent': 4.20, 'battery': 13.45, 'batSOC': 75, 'batTemperature': 25.0, 'batStatut': 1, 'batSousVoltage': False, 'batSurVoltage': False, 'loadVoltage': 13.40, 'loadCurrent': 2.00, 'loadPower': 26.80, 'energieGenJour': 1.50, 'energieGenTotal': 120.00, 'energieConJour': 0.80, 'energieConTotal': 85.00, 'sun': True, 'relay1': False, 'relay2': False, 'di1': False, 'di2': False, 'autoMode': True, 'rs485_ok': True, 'last_update': 0, } rules = [] next_id = [1] sleep_cfg = {'actif': False, 'intervalle': 600, 'seuil': 2.0} # Historique circulaire (échantillonnage toutes les 5s en simulation) history = {'b': [], 'p': [], 'l': [], 's': []} MAX_HIST = 288 def _update(): """Mise à jour périodique de l'état et de l'historique.""" while True: t = time.time() with _lock: state['pv'] = round(18.72 + 0.8 * math.sin(t / 60), 2) state['pvCurrent'] = round(max(0, 4.20 + 0.4 * math.sin(t / 45)), 2) state['battery'] = round(13.45 + 0.3 * math.sin(t / 120), 2) state['loadPower'] = round(max(0, 26.80 + 3.0 * math.sin(t / 30)), 1) state['batSOC'] = min(100, max(0, int(75 + 5 * math.sin(t / 180)))) state['sun'] = (int(t / 30) % 2) == 0 # alterne jour/nuit toutes les 30s state['rs485_ok'] = True state['last_update'] = int(t * 1000) # Historique — 1 point toutes les 5s (= 5 min en temps réel) h = history if len(h['b']) >= MAX_HIST: for k in h: h[k].pop(0) h['b'].append(round(state['battery'], 2)) h['p'].append(round(state['pv'], 2)) h['l'].append(round(state['loadPower'], 1)) h['s'].append(state['batSOC']) time.sleep(5) threading.Thread(target=_update, daemon=True).start() # --------------------------------------------------------------------------- # Handler HTTP # --------------------------------------------------------------------------- class Handler(http.server.BaseHTTPRequestHandler): def do_OPTIONS(self): self._cors(200) def do_GET(self): p = urlparse(self.path) if p.path == '/api/state': with _lock: self._json(state) elif p.path == '/api/rules': with _lock: self._json(rules) elif p.path == '/api/sleep': with _lock: self._json(sleep_cfg) elif p.path == '/api/history': with _lock: out = {'n': len(history['b'])} out.update(history) self._json(out) else: self._static(p.path) def do_POST(self): p = urlparse(self.path) qs = parse_qs(p.query) body = self._read_body() # --- Relais --- if p.path.startswith('/api/relay/'): parts = p.path.split('/') # ['','api','relay','1','on'] n, cmd = int(parts[3]), parts[4] key = f'relay{n}' with _lock: state[key] = (cmd == 'on') self._json({'ok': True}) # --- Mode --- elif p.path == '/api/mode/auto': with _lock: state['autoMode'] = True self._json({'ok': True}) elif p.path == '/api/mode/manuel': with _lock: state['autoMode'] = False self._json({'ok': True}) # --- Règles --- elif p.path == '/api/rules' and body: try: r = json.loads(body) with _lock: r['id'] = next_id[0]; next_id[0] += 1 rules.append(r) self._json({'ok': True}, 201) except Exception: self._json({'ok': False}, 400) elif p.path == '/api/rules/toggle': rid = int(qs.get('id', [0])[0]) with _lock: found = next((r for r in rules if r['id'] == rid), None) if found: found['enabled'] = not found['enabled'] self._json({'ok': bool(found)}, 200 if found else 404) elif p.path == '/api/rules/delete': rid = int(qs.get('id', [0])[0]) with _lock: before = len(rules) rules[:] = [r for r in rules if r['id'] != rid] ok = len(rules) < before self._json({'ok': ok}, 200 if ok else 404) # --- Sleep --- elif p.path == '/api/sleep' and body: try: cfg = json.loads(body) with _lock: sleep_cfg.update({ 'actif': bool(cfg.get('actif', sleep_cfg['actif'])), 'intervalle': int(cfg.get('intervalle', sleep_cfg['intervalle'])), 'seuil': float(cfg.get('seuil', sleep_cfg['seuil'])), }) self._json({'ok': True}) except Exception: self._json({'ok': False}, 400) else: self.send_error(404) # ------------------------------------------------------------------ def _static(self, path): if path in ('', '/'): path = '/index.html' filepath = os.path.normpath(os.path.join(DATA_DIR, path.lstrip('/'))) # Sécurité : rester dans DATA_DIR if not filepath.startswith(os.path.realpath(DATA_DIR)): self.send_error(403); return if not os.path.isfile(filepath): self.send_error(404); return ext = os.path.splitext(filepath)[1] with open(filepath, 'rb') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', MIME.get(ext, 'application/octet-stream')) self.send_header('Content-Length', str(len(data))) self._cors_headers() self.end_headers() self.wfile.write(data) def _json(self, obj, code=200): data = json.dumps(obj).encode() self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(data))) self._cors_headers() self.end_headers() self.wfile.write(data) def _cors(self, code): self.send_response(code) self._cors_headers() self.end_headers() def _cors_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') def _read_body(self): length = int(self.headers.get('Content-Length', 0)) return self.rfile.read(length).decode() if length else '' def log_message(self, fmt, *args): print(f'[Sim] {self.address_string()} {fmt % args}', flush=True) # --------------------------------------------------------------------------- if __name__ == '__main__': data_real = os.path.realpath(DATA_DIR) if not os.path.isdir(data_real): print(f'[Sim] ERREUR : dossier data introuvable : {data_real}') raise SystemExit(1) with socketserver.ThreadingTCPServer(('', PORT), Handler) as httpd: httpd.allow_reuse_address = True print(f'[Sim] Serveur de simulation sur http://localhost:{PORT}') print(f'[Sim] Fichiers web depuis : {data_real}') print(f'[Sim] Historique : 1 point / 5s (288 pts max = ~24 min simulées)') httpd.serve_forever()