From 6fcce1a470156ff5886391a5c70d666d62bffd8c Mon Sep 17 00:00:00 2001 From: Patrick Schurig Date: Sat, 17 Oct 2026 13:16:43 +0200 Subject: [PATCH] =?UTF-8?q?feat(tools):=20etm-powersync-ref-optimizer,=20o?= =?UTF-8?q?ptimiseur=20de=20r=C3=A9f=C3=A9rence=201.1=20(brief=204=20T7,?= =?UTF-8?q?=20annexe=20E)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python 3, bibliothèque standard seule, GPL-3.0. Socket unix par défaut (unix:///run/powersync/optimizer.sock, 0660), tcp:// accepté. Handshake (majeure, capacités, token lu dans un fichier et relu à chaque handshake, -32001 puis fermeture s'il diffère), GetPlan, abstention, plusieurs clients à la fois (un fil par connexion, plan fonction pure de la requête). Plan à un créneau qui reproduit la passe confort du waterfall rule-based depuis le seul contexte GetPlan : tri (rang, id), surplus net signé, recrédit, réserve de stockage, plancher, plafond, verrous, mapping SG-Ready avec hystérésis. Codes du catalogue émissible seulement, clés documentées seulement (--list-codes, confronté au moteur par le test). Modes d'injection : normal, abstain, slow, invalid, expired, wrong-major, drop, unknown-load, forbidden-kind, missing-load, echo-token, protocol-1.0, deviate ; --mode-file pour changer de mode sans reconnexion. README. Co-Authored-By: Claude Opus 5.5 --- tools/etm-powersync-ref-optimizer/README.md | 67 +++ .../etm-powersync-ref-optimizer | 526 ++++++++++++++++++ 2 files changed, 593 insertions(+) create mode 100644 tools/etm-powersync-ref-optimizer/README.md create mode 100755 tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer diff --git a/tools/etm-powersync-ref-optimizer/README.md b/tools/etm-powersync-ref-optimizer/README.md new file mode 100644 index 0000000..ab342b6 --- /dev/null +++ b/tools/etm-powersync-ref-optimizer/README.md @@ -0,0 +1,67 @@ +# `etm-powersync-ref-optimizer` — optimiseur de référence + +Python 3, bibliothèque standard seule, GPL-3.0 comme le dépôt. Protocole : `docs/OPTIMIZER_PROTOCOL.md` +1.1 (annexe E). Il répond un plan à **un créneau** qui reproduit le waterfall du rule-based (passe +confort : même ordre, même budget, mêmes écrêtages) en ne lisant **que** le contexte `GetPlan` — aucun +accès réseau sortant, aucun fichier hors le token. Usages : test de bout en bout du moteur, point de +comparaison d'un banc, exemple pour qui écrit son propre optimiseur. + +Ce qu'il ne reproduit pas (le contexte 1.1 ne le porte pas) : passe éco, dérogations, plafond de +soutirage, paliers d'un `relay-router` (il envoie une consigne continue que le routeur arrondit), +refus de commande, proxy VE. L'en-tête du script le détaille. + +## Lancer + +```sh +# socket local (défaut : unix:///run/powersync/optimizer.sock, mode 0660) +./tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer + +# sur le LAN, avec un token (lu dans un FICHIER, jamais en argument) +./tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer --listen tcp://0.0.0.0:7600 \ + --token-file /etc/powersync/optimizer.token +``` + +Il écrit `ÉCOUTE ` sur la sortie standard quand il écoute (le port réel de `tcp://…:0`), +et son journal sur la sortie d'erreur. Plusieurs clients à la fois (un fil par connexion, aucun état +partagé) : le moteur et un rejeu peuvent lui parler ensemble. SIGTERM l'arrête proprement. + +## Pointer le moteur dessus + +```sh +# méthodes NymeaEnergy (permission Admin) — INTERFACE.md, « Optimiseur » +NymeaEnergy.SetOptimizerEndpoint {"endpoint": "unix:///run/powersync/optimizer.sock"} +NymeaEnergy.SetOptimizerToken {"token": ""} # clé seule +NymeaEnergy.GetOptimizerStatus # optimizerAlive, activeStrategy, name, lastErrorCode… +``` + +Sur `tcp://` sans TLS, le token circule en clair sur le LAN (protocole §3). + +## Traces, puis rejeu + +```sh +NymeaEnergy.SetOptimizerTrace {"enabled": true} # défaut : désactivé +NymeaEnergy.GetOptimizerTrace # → directory +# une ligne par cycle : /etm-powersync-optimizer-trace-AAAA-MM-JJ.ndjson + +# rejouer une journée sur un optimiseur (celui-ci, ou un candidat) +./tools/etm-powersync-trace-replay /etm-powersync-optimizer-trace-2026-10-17.ndjson \ + --endpoint unix:///run/powersync/optimizer.sock --fail-on-divergence +``` + +Le rejeu compare, cycle par cycle, les actions du créneau courant au plan enregistré (`tools/README.md`). + +## Modes d'injection de défauts (`--mode`, tests) + +`normal` · `abstain` · `slow` (répond après `--slow-s`, défaut 12 s, au-delà des 10 s du moteur) · +`invalid` (`slots` n'est pas une liste) · `expired` (`validUntil` passé) · `wrong-major` (handshake +2.0) · `drop` (ferme la connexion à chaque `GetPlan`) · `unknown-load` · `forbidden-kind` (kind +`constraint`) · `missing-load` (une charge disponible absente du créneau courant) · `echo-token` +(recopie le token dans `reason` et `planId`) · `protocol-1.0` (optimiseur 1.0 : sans champs 1.1) · +`deviate` (cascade en ordre inverse : doit diverger). + +`--mode-file FICHIER` : le mode est relu dans ce fichier à chaque message — un test change de mode +sans reconnexion. `--list-codes` : le catalogue des codes émissibles qu'il embarque (le test +d'intégration le confronte au moteur). + +Tests : `tests/auto/refoptimizer` (suite `etm-powersync-refoptimizer`, moteur réel + ce script en +sous-processus). diff --git a/tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer b/tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer new file mode 100755 index 0000000..02b8701 --- /dev/null +++ b/tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer @@ -0,0 +1,526 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-3.0-or-later +# Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync +"""Optimiseur de référence (OPTIMIZER_PROTOCOL.md 1.1, annexe E). + + etm-powersync-ref-optimizer [--listen unix:///run/powersync/optimizer.sock | tcp://hôte:port] + [--token-file FICHIER] [--mode MODE | --mode-file FICHIER] + [--slow-s N] [--horizon-s S] + etm-powersync-ref-optimizer --list-codes + +Un serveur JSON-RPC 2.0 sur NDJSON qui répond au `Handshake` et à `GetPlan` par un plan à UN +créneau reproduisant le waterfall du rule-based — même ordre (rang, puis identifiant), même +budget (surplus net signé, recrédit de la consommation courante, réserve de stockage), mêmes +écrêtages (plancher, plafond, verrous, fenêtre SG-Ready). Il ne lit QUE le contexte `GetPlan` : +aucun accès réseau sortant, aucun fichier hors le token. + +Ce qu'il ne reproduit PAS, faute de champ dans le contexte 1.1 (écarts du brief 4 T7) : la passe +éco (avancement livré absent), les dérogations, l'autorisation de soutirage, l'arrondi sur les +paliers d'un `relay-router` (`powerLevels` absent, E9), le refus de commande, et le proxy VE +(échéance, prix, tolérance d'acquisition). Il est donc identique au rule-based sur la passe +confort seule (décision Q6), et c'est ce que le test d'intégration éprouve. + +Modes (`--mode`, pour les tests ; un seul à la fois) : + normal plan du waterfall + abstain {"abstain": true} + slow plan du waterfall, après --slow-s secondes (défaut 12, au-delà du délai de 10 s) + invalid plan mal formé (`slots` n'est pas une liste) + expired `validUntil` une minute AVANT l'instant de la requête + wrong-major handshake en 2.0 (majeure refusée par le moteur) + drop ferme la connexion à chaque GetPlan, sans répondre + unknown-load une action de plus, sur une charge absente du contexte + forbidden-kind la première action en kind `constraint` (réservé, annexe F) + missing-load la première charge disponible absente du créneau courant (Q1 : invalide) + echo-token le token recopié dans `reason` et `planId` (test du masquage, correctif A) + protocol-1.0 un optimiseur 1.0 : handshake 1.0 sans `capabilities`, plan sans `validUntil`, + `reasonCode`, `reasonParams` ni `diagnostics` (§9 : reste accepté) + deviate cascade servie en ordre de rang INVERSE (contre-épreuve : doit diverger) + +`--mode-file` : le mode est relu dans ce fichier à chaque message (un mot ; absent ou illisible = +`normal`). Un test change ainsi de mode SANS reconnexion — ce qui distingue « abstention, optimiseur +toujours vivant » de « socket fermé ». + +Plusieurs clients à la fois : chaque connexion a son fil, et un plan ne dépend que de sa requête +(aucun état partagé). Le moteur et un rejeu (`etm-powersync-trace-replay`) peuvent donc parler au +même optimiseur, et une connexion morte n'en bloque pas une neuve. + +Le token se lit dans un FICHIER (--token-file), relu à chaque handshake, jamais en argument (il +se verrait dans `ps`) et jamais affiché. S'il est configuré, un handshake sans lui, ou avec un +autre, reçoit une erreur JSON-RPC -32001 puis la connexion est fermée. + +Sur la sortie standard, une seule ligne : `ÉCOUTE ` quand le serveur écoute (le port +réel d'un `tcp://…:0`). Le journal va sur la sortie d'erreur. +""" + +import argparse +import json +import math +import os +import signal +import socket +import socketserver +import stat +import sys +import threading +import time +from datetime import datetime, timedelta + +ENDPOINT_DEFAUT = 'unix:///run/powersync/optimizer.sock' +VERSION = '1.1' +NOM = 'etm-powersync-ref-optimizer' +LIGNE_MAX = 1024 * 1024 # E-T2-1 : une ligne de plus de 1 Mio est écartée +CODE_TOKEN_REFUSE = -32001 # E-T2-6 + +MODES = ['normal', 'abstain', 'slow', 'invalid', 'expired', 'wrong-major', 'drop', 'unknown-load', + 'forbidden-kind', 'missing-load', 'echo-token', 'protocol-1.0', 'deviate'] + +# Catalogue (energyplugin/etm/types/decisionreason.cpp) : codes qu'un optimiseur peut émettre et +# leurs clés de paramètres documentées (INTERFACE.md). Le test d'intégration les confronte au +# moteur (`--list-codes`) : une divergence le fait échouer. +CODES_EMISSIBLES = { + 'SURPLUS_SETPOINT': ['budgetW', 'setpointW', 'stepped'], + 'SURPLUS_INSUFFICIENT': ['budgetW'], + 'BELOW_MIN_POWER': ['budgetW', 'minPowerW', 'state'], + 'EV_DEADLINE': [], + 'SG_FORCED': ['budgetW', 'estimatedW'], + 'SG_RECOMMENDED': ['budgetW', 'estimatedW'], + 'SG_NORMAL': ['budgetW'], + 'ECO_FLOOR_GRID': ['gridW', 'atMaximum', 'remainingWh', 'targetWh', 'remainingS', 'regime', + 'minAlwaysW', 'boundBy'], + 'PRICE_SLOT': ['priceCtkWh', 'setpointW'], + 'OPTIMIZER_DECISION': [], +} + +# Hystérésis d'entrée en état 4 du rule-based (rulebasedscheduler.cpp, kForceMargin). +MARGE_FORCE = 1.2 + + +def journal(message): + print(f'[{NOM}] {message}', file=sys.stderr, flush=True) + + +def q_round(x): + """qRound de Qt : moitié arrondie en s'éloignant de zéro (round() de Python arrondit au pair).""" + return int(math.floor(x + 0.5)) if x >= 0 else int(math.ceil(x - 0.5)) + + +def nombre(v, defaut=0.0): + return float(v) if isinstance(v, (int, float)) and not isinstance(v, bool) else defaut + + +# --- Le waterfall (passe confort du rule-based) ------------------------------------------------ + +def motif(code, params, texte): + """Une action porte un code du catalogue, ses seules clés documentées, et un texte.""" + if code not in CODES_EMISSIBLES: + # Le rule-based a un code que l'optimiseur n'a pas le droit d'émettre (verrou, réserve, + # indisponibilité…) : le code générique, le nom du mécanisme dans le texte seulement. + return {'reason': f'{texte} ({code})', 'reasonCode': 'OPTIMIZER_DECISION'} + cles = CODES_EMISSIBLES[code] + out = {'reason': texte, 'reasonCode': code} + p = {k: v for k, v in params.items() if k in cles} + if p: + out['reasonParams'] = p + return out + + +def budget_initial(ctx): + """Surplus net signé, puis la règle de stockage — rulebasedscheduler.cpp, getPlan().""" + meter = ctx.get('meter') or {} + restant = nombre(meter.get('exportW')) - nombre(meter.get('importW')) + reserve = {'active': False, 'socPercent': 0.0, 'seuil': 0.0, 'retenuW': 0.0} + bat = ctx.get('battery') or {} + if bat.get('present') is True: + soc = nombre(bat.get('socPercent')) + seuil = nombre(bat.get('reserveSocPercent')) + puissance = nombre(bat.get('powerW')) + if soc < seuil: + reserve.update(active=True, socPercent=soc, seuil=seuil, retenuW=max(0.0, restant)) + restant = min(restant, 0.0) + if puissance > 0: + restant -= 500 + elif puissance > 0: + restant += puissance + if puissance < 0: + restant += puissance + return restant, reserve + + +def consigne(lc, restant, reserve): + """decideSetpoint + choisirMotif pour une charge en watts. Rend (action ou None, restant).""" + decl = lc.get('declared') or {} + tele = lc.get('telemetry') or {} + if lc.get('adapter') == 'evcharger': + # Approximation : plancher et plafond déduits de la plaque (le moteur les recalcule à + # chaque cycle avec la limite de phase — écart E26). Non éprouvé (pas de borne au banc). + phases = max(1.0, nombre(decl.get('phases'), 1.0)) + min_w = nombre(decl.get('minA')) * 230.0 * phases + max_w = nombre(decl.get('maxA')) * 230.0 * phases + else: + min_w = nombre(decl.get('minPowerW')) + max_w = nombre(decl.get('maxPowerW')) + courant = nombre(tele.get('currentPowerW')) + recredit = 0.0 if reserve['active'] else courant + budget = restant + recredit + plafond = max_w if max_w > 0 else budget + + if min_w > 0 and plafond < min_w: + action = {'loadId': lc['id'], 'kind': 'setpoint', 'powerW': 0.0} + action.update(motif('PHASE_LIMIT', {}, f'plafond {q_round(plafond)} W sous le plancher {q_round(min_w)} W')) + return action, budget + + consigne_w = max(0.0, min(budget, plafond)) # qBound(0, budget, plafond) + sous_plancher, plancher = False, 0.0 + if min_w > 0 and 0 < consigne_w < min_w: + consigne_w, sous_plancher, plancher = 0.0, True, min_w + + if tele.get('available') is False: + # Indisponible : hors de l'arbitrage, pas de la comptabilité — sa puissance figée reste + # retranchée. Aucune action (Q1 : une charge indisponible n'a pas à figurer au créneau). + return None, budget - nombre(tele.get('lockMinPowerW')) + + voulu = consigne_w + lock_max = nombre(tele.get('lockMaxPowerW'), -1.0) + lock_min = nombre(tele.get('lockMinPowerW')) + if lock_max >= 0: + consigne_w = min(consigne_w, lock_max) + if consigne_w < lock_min: + consigne_w = lock_min + + if consigne_w > voulu: + m = motif('LOCK_MIN_ON', {}, f'maintenue à {q_round(consigne_w)} W par le verrou de marche minimale') + elif consigne_w < voulu: + m = motif('LOCK_MIN_OFF', {}, f'retenue à {q_round(consigne_w)} W par le verrou d\'arrêt minimal') + elif consigne_w > 0: + m = motif('SURPLUS_SETPOINT', {'budgetW': q_round(budget), 'setpointW': q_round(consigne_w), + 'stepped': lc.get('adapter') == 'relay-router'}, + f'surplus {q_round(budget)} W → {q_round(consigne_w)} W') + elif reserve['active'] and reserve['retenuW'] + courant > 0: + m = motif('BATTERY_RESERVE', {}, f'stockage sous sa réserve ({q_round(reserve["socPercent"])} %)') + elif sous_plancher: + m = motif('BELOW_MIN_POWER', {'budgetW': q_round(budget), 'minPowerW': q_round(plancher)}, + f'budget {q_round(budget)} W sous le plancher {q_round(plancher)} W') + else: + m = motif('SURPLUS_INSUFFICIENT', {'budgetW': q_round(budget)}, f'surplus insuffisant ({q_round(budget)} W)') + action = {'loadId': lc['id'], 'kind': 'setpoint', 'powerW': consigne_w} + action.update(m) + return action, budget - consigne_w + + +def etat_sg(lc, restant, reserve): + """buildSgReadyStateAction : mapping sémantique, hystérésis, fenêtre de verrou.""" + decl = lc.get('declared') or {} + tele = lc.get('telemetry') or {} + estimes = {} + for k, v in (decl.get('estimatedPowerW') or {}).items(): + try: + estimes[int(k)] = nombre(v) + except ValueError: + pass + etats = sorted(int(s) for s in decl.get('states') or [] if isinstance(s, int)) + courant = int(nombre(tele.get('state'))) + p3, p4 = estimes.get(3, 0.0), estimes.get(4, 0.0) + budget = restant + (0.0 if reserve['active'] else estimes.get(courant, 0.0)) + + if p4 > 0 and budget >= p4 * MARGE_FORCE: + cible = 4 + elif courant == 4 and p4 > 0 and budget >= p4: + cible = 4 + elif p3 > 0 and budget >= p3: + cible = 3 + else: + cible = 2 + bas = int(nombre(tele.get('minState'))) or 2 + haut = int(nombre(tele.get('maxState'))) or 2 + retenu = min(max(cible, min(bas, haut)), max(bas, haut)) + if retenu not in etats: + cale = etats[0] if etats else 2 + for s in etats: + if cale < s <= retenu: + cale = s + retenu = cale + puissance = estimes.get(retenu, 0.0) + + if retenu != cible: + m = motif('LOCK_MIN_STATE_HOLD', {}, f'état {retenu} tenu par le verrou d\'état') + elif retenu == 4: + m = motif('SG_FORCED', {'budgetW': q_round(budget), 'estimatedW': q_round(p4)}, f'surplus {q_round(budget)} W → forcé') + elif retenu == 3: + m = motif('SG_RECOMMENDED', {'budgetW': q_round(budget), 'estimatedW': q_round(p3)}, + f'surplus {q_round(budget)} W → recommandation') + elif reserve['active'] and reserve['retenuW'] > 0: + m = motif('BATTERY_RESERVE', {}, f'stockage sous sa réserve ({q_round(reserve["socPercent"])} %)') + elif budget > 0 and p3 > 0: + m = motif('BELOW_MIN_POWER', {'budgetW': q_round(budget), 'minPowerW': q_round(p3), 'state': 3}, + f'budget {q_round(budget)} W sous l\'état 3 ({q_round(p3)} W)') + else: + m = motif('SG_NORMAL', {'budgetW': q_round(budget)}, f'pas de surplus ({q_round(budget)} W) → normal') + if tele.get('available') is False: + return None, budget - puissance + action = {'loadId': lc['id'], 'kind': 'state', 'state': retenu} + action.update(m) + return action, budget - puissance + + +ARBITREES = ('etmvariableload', 'relay-router', 'sg-ready', 'evcharger') + + +def waterfall(ctx, inverse=False): + """Les actions du créneau courant, dans l'ordre de service.""" + restant, reserve = budget_initial(ctx) + charges = [lc for lc in ctx.get('loads') or [] + if isinstance(lc, dict) and isinstance(lc.get('id'), str) and lc.get('adapter') in ARBITREES] + charges.sort(key=lambda lc: (nombre(lc.get('priority')), lc['id']), reverse=inverse) + actions = [] + for lc in charges: + if lc.get('adapter') == 'sg-ready': + action, restant = etat_sg(lc, restant, reserve) + else: + action, restant = consigne(lc, restant, reserve) + if action is not None: + actions.append(action) + return actions + + +def instant(texte): + if not isinstance(texte, str): + return None + try: + t = datetime.fromisoformat(texte.replace('Z', '+00:00')) + except ValueError: + return None + return t if t.tzinfo is not None else None + + +def iso(t): + return t.isoformat(timespec='seconds') + + +def plan(ctx, mode, horizon_s, token): + """La réponse à un GetPlan (`result`), ou None pour « ne pas répondre » (drop).""" + t0 = time.monotonic() + maintenant = instant(ctx.get('timestamp')) + if maintenant is None: + return {'abstain': True, 'reason': 'timestamp illisible ou sans décalage'} + if mode == 'abstain': + return {'abstain': True, 'reason': 'mode abstain (test)'} + actions = waterfall(ctx, inverse=(mode == 'deviate')) + if mode == 'unknown-load': + actions.append({'loadId': 'charge-inconnue-du-contexte', 'kind': 'setpoint', 'powerW': 0.0, + 'reason': 'mode unknown-load (test)', 'reasonCode': 'OPTIMIZER_DECISION'}) + elif mode == 'forbidden-kind' and actions: + actions[0] = dict(actions[0], kind='constraint', charge='allow') + actions[0].pop('powerW', None) + actions[0].pop('state', None) + elif mode == 'missing-load': + disponibles = {lc.get('id') for lc in ctx.get('loads') or [] + if isinstance(lc, dict) and (lc.get('telemetry') or {}).get('available') is not False} + for i, a in enumerate(actions): + if a['loadId'] in disponibles: + del actions[i] + break + elif mode == 'echo-token' and token: + for a in actions: + a['reason'] = f'{a["reason"]} — écho {token}' + fin = maintenant + timedelta(seconds=horizon_s) + validite = maintenant - timedelta(seconds=60) if mode == 'expired' else fin + resultat = { + 'planId': f'ref-{iso(maintenant)}' + (f'-{token}' if mode == 'echo-token' and token else ''), + 'strategy': 'ref-waterfall', + 'computedAt': iso(maintenant), + 'validUntil': iso(validite), + 'slots': [{'from': iso(maintenant), 'to': iso(fin), 'actions': actions}], + 'diagnostics': {'status': 'optimal', 'computeMs': q_round((time.monotonic() - t0) * 1000), + 'inputsUsed': {'pv': 'none'}}, + } + if mode == 'invalid': + resultat['slots'] = 'mode invalid (test) : slots n\'est pas une liste' + elif mode == 'protocol-1.0': + # Ce qu'un optimiseur 1.0 savait dire : identifiant, stratégie, créneaux, actions et texte. + for champ in ('validUntil', 'computedAt', 'diagnostics'): + del resultat[champ] + for a in actions: + a.pop('reasonCode', None) + a.pop('reasonParams', None) + return resultat + + +def mode_courant(cfg): + if not cfg.mode_file: + return cfg.mode + try: + with open(cfg.mode_file, encoding='utf-8') as f: + mot = f.read(64).strip() + except OSError: + return 'normal' + return mot if mot in MODES else 'normal' + + +# --- Transport ----------------------------------------------------------------------------------- + +def lire_token(chemin): + if not chemin: + return None + try: + with open(chemin, 'rb') as f: + return f.read(8192).strip().decode('utf-8', 'replace') or None + except OSError as e: + journal(f'fichier de token illisible ({e.strerror}) — tout handshake refusé') + return '' # configuré mais illisible : rien ne correspond + + +class Connexion(socketserver.StreamRequestHandler): + """Une connexion = un fil ; requêtes traitées dans l'ordre d'arrivée.""" + + def envoyer(self, message): + ligne = json.dumps(message, separators=(',', ':'), ensure_ascii=False).encode() + b'\n' + self.wfile.write(ligne) + self.wfile.flush() + + def handle(self): + cfg = self.server.cfg + pret = False + token = None + while True: + try: + ligne = self.rfile.readline(LIGNE_MAX + 2) + except (OSError, ValueError): + return + if not ligne: + return + if len(ligne) > LIGNE_MAX: + journal('ligne trop longue écartée') + continue + try: + message = json.loads(ligne) + except ValueError: + journal('ligne non JSON écartée') + continue + if not isinstance(message, dict) or 'method' not in message: + continue + ident, methode = message.get('id'), message.get('method') + params = message.get('params') if isinstance(message.get('params'), dict) else {} + if ident is None: + continue # notification du moteur : aucune n'est définie + mode = mode_courant(cfg) + if methode == 'Handshake': + attendu = lire_token(cfg.token_file) + if attendu is not None and params.get('token') != attendu: + # Le token reçu n'est jamais recopié ni journalisé. + journal('handshake refusé : token absent ou différent') + self.envoyer({'jsonrpc': '2.0', 'id': ident, + 'error': {'code': CODE_TOKEN_REFUSE, 'message': 'token refusé'}}) + return + token = attendu + pret = True + reponse = {'protocolVersion': VERSION, 'name': NOM, 'planning': False, + 'capabilities': {'resolutionS': cfg.horizon_s, 'kinds': ['setpoint', 'state'], + 'usesForecast': [], 'reasonCodes': True}} + if mode == 'wrong-major': + reponse['protocolVersion'] = '2.0' + elif mode == 'protocol-1.0': + reponse['protocolVersion'] = '1.0' + del reponse['capabilities'] + self.envoyer({'jsonrpc': '2.0', 'id': ident, 'result': reponse}) + journal(f'handshake accepté ({str(params.get("engine", "?"))[:80]}), mode {mode}') + elif methode == 'GetPlan': + if not pret: + self.envoyer({'jsonrpc': '2.0', 'id': ident, + 'error': {'code': -32002, 'message': 'handshake attendu'}}) + continue + if mode == 'drop': + journal('mode drop : connexion fermée sans réponse') + return + if mode == 'slow': + time.sleep(cfg.slow_s) + self.envoyer({'jsonrpc': '2.0', 'id': ident, + 'result': plan(params, mode, cfg.horizon_s, token)}) + else: + self.envoyer({'jsonrpc': '2.0', 'id': ident, + 'error': {'code': -32601, 'message': 'méthode inconnue'}}) + + +class ServeurUnix(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): + daemon_threads = True + + +class ServeurTcp(socketserver.ThreadingMixIn, socketserver.TCPServer): + daemon_threads = True + allow_reuse_address = True + + +class ServeurTcp6(ServeurTcp): + address_family = socket.AF_INET6 + + +def ecouter(endpoint): + if endpoint.startswith('unix://'): + chemin = endpoint[len('unix://'):] + if not chemin.startswith('/'): + raise SystemExit(f'endpoint unix sans chemin absolu : {endpoint}') + if os.path.exists(chemin): + if not stat.S_ISSOCK(os.lstat(chemin).st_mode): + raise SystemExit(f'{chemin} existe et n\'est pas un socket : refus de l\'écraser') + os.unlink(chemin) # socket orphelin d'une exécution précédente + serveur = ServeurUnix(chemin, Connexion) + os.chmod(chemin, 0o660) # §2 : 0660 (le groupe `powersync` se pose au déploiement) + return serveur, endpoint + if endpoint.startswith('tcp://'): + reste = endpoint[len('tcp://'):].rstrip('/') + if reste.startswith('['): + hote, _, port = reste[1:].partition(']:') + classe = ServeurTcp6 + else: + hote, _, port = reste.rpartition(':') + classe = ServeurTcp + if not hote or not port.isdigit(): + raise SystemExit(f'endpoint tcp illisible (tcp://hôte:port attendu) : {endpoint}') + serveur = classe((hote, int(port)), Connexion) + reel = serveur.server_address[1] + return serveur, f'tcp://[{hote}]:{reel}' if classe is ServeurTcp6 else f'tcp://{hote}:{reel}' + raise SystemExit(f'endpoint inconnu (unix:///chemin ou tcp://hôte:port) : {endpoint}') + + +def main(argv=None): + p = argparse.ArgumentParser(prog=NOM, description='Optimiseur de référence (protocole 1.1, annexe E).') + p.add_argument('--listen', default=ENDPOINT_DEFAUT, + help=f'unix:///chemin ou tcp://hôte:port (défaut : {ENDPOINT_DEFAUT})') + p.add_argument('--token-file', help='fichier contenant le token attendu (jamais en argument)') + p.add_argument('--mode', choices=MODES, default='normal', help='injection de défaut (défaut : normal)') + p.add_argument('--mode-file', help='fichier relu à chaque message, qui porte le mode (tests)') + p.add_argument('--slow-s', type=float, default=12.0, help='délai du mode slow, en s (défaut 12)') + p.add_argument('--horizon-s', type=int, default=300, + help='durée du créneau et de validUntil, en s (défaut 300)') + p.add_argument('--list-codes', action='store_true', + help='écrit le catalogue des codes émissibles (JSON) et sort') + args = p.parse_args(argv) + if args.list_codes: + print(json.dumps(CODES_EMISSIBLES, sort_keys=True)) + return 0 + def arreter(*_): + raise KeyboardInterrupt + signal.signal(signal.SIGTERM, arreter) # arrêt propre : le socket unix est retiré + serveur, reel = ecouter(args.listen) + serveur.cfg = args + print(f'ÉCOUTE {reel}', flush=True) + journal(f'écoute sur {reel}, mode {args.mode_file and "relu dans " + args.mode_file or args.mode}, token {"attendu" if args.token_file else "non configuré"}') + fil = threading.Thread(target=serveur.serve_forever, daemon=True) + fil.start() + try: + while fil.is_alive(): + fil.join(0.5) + except KeyboardInterrupt: + pass + finally: + serveur.shutdown() + serveur.server_close() + if reel.startswith('unix://'): + try: + os.unlink(reel[len('unix://'):]) + except OSError: + pass + return 0 + + +if __name__ == '__main__': + sys.exit(main())