feat(tools): etm-powersync-ref-optimizer, optimiseur de référence 1.1 (brief 4 T7, annexe E)
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 <noreply@anthropic.com>
This commit is contained in:
Patrick Schurig
parent
693e01579f
commit
6fcce1a470
67
tools/etm-powersync-ref-optimizer/README.md
Normal file
67
tools/etm-powersync-ref-optimizer/README.md
Normal file
@ -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 <endpoint>` 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": "<le même secret que dans --token-file>"} # 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 : <directory>/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 <directory>/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).
|
||||
526
tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer
Executable file
526
tools/etm-powersync-ref-optimizer/etm-powersync-ref-optimizer
Executable file
@ -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 <endpoint>` 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())
|
||||
Loading…
x
Reference in New Issue
Block a user