4cbbb81802
Il s'annonce 1.2 et s'aligne sur les règles écrites par ce lot :
- Q3 : GetPlan avant un handshake réussi → -32600 (était -32002, hors norme),
connexion gardée ; Handshake d'une autre majeure → -32602 puis fermeture
(il ne comparait pas la majeure du moteur) ; token refusé → -32001, inchangé.
- Q2 : un timestamp illisible ou sans décalage n'est plus une abstention mais
-32602, error.data.problems = [{path, problem}], jamais la valeur reçue.
- Q7 : une ligne de plus de 1 Mio ('\n' non compté) reçoit -32600 à id null.
Correctif au passage : readline(LIGNE_MAX + 2) rendait le reste d'une
ligne trop longue comme des lignes suivantes, et comptait le '\n' (une
ligne d'exactement 1 Mio était refusée). La ligne est désormais lue jusqu'à
sa fin par morceaux, sans être gardée.
- Q8 : un lot reçoit -32600 à id null, aucun de ses messages n'est traité
(il était ignoré en silence).
- -32700 à id null pour une ligne non JSON (ignorée en silence avant).
- Q5 et Q6 étaient déjà tenus : token accepté sans --token-file,
UpdateLearned en notification ignorée, avec id → -32601.
README : tableau des erreurs.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
628 lines
29 KiB
Python
Executable File
628 lines
29 KiB
Python
Executable File
#!/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.2, 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] [--capture FICHIER]
|
|
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)
|
|
wrong-type `planId` numérique au lieu d'une chaîne : le moteur l'ACCEPTE (il le lit comme
|
|
texte), le contrat OpenRPC le REFUSE — contre-épreuve de la suite de conformité
|
|
|
|
`--capture FICHIER` : chaque message reçu et envoyé, une ligne JSON-RPC par message, dans l'ordre
|
|
(la session que `tools/etm-powersync-schema-check --ndjson` juge). Le token d'un handshake y est
|
|
remplacé par autant d'astérisques : sa FORME reste vérifiable, sa valeur n'est jamais écrite.
|
|
|
|
`--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. Sans --token-file, un
|
|
token reçu est accepté sans être lu.
|
|
|
|
Erreurs (protocole 1.2, §2, §3, §7 ; codes JSON-RPC 2.0, plus -32001) :
|
|
Handshake d'une autre majeure -32602, puis fermeture (comme -32001 : les deux côtés ferment)
|
|
GetPlan avant un handshake réussi -32600, connexion gardée
|
|
contexte GetPlan invalide -32602, `error.data.problems` = [{path, problem}], jamais la
|
|
valeur reçue — une erreur, pas une abstention
|
|
ligne non JSON -32700, id null
|
|
lot JSON-RPC (`[…]`) -32600, id null, aucun message du lot traité
|
|
ligne de plus de 1 Mio -32600, id null, jetée sans être gardée en mémoire
|
|
requête d'une méthode non servie -32601 (dont UpdateLearned, qui ne va que vers le moteur) ;
|
|
une notification (sans id) est ignorée
|
|
Hors handshake refusé, la connexion est toujours gardée.
|
|
|
|
Au handshake, il s'annonce `powersync-optimizer` (exemple de référence GPL) ; `etm-optimizer` est
|
|
l'optimiseur ETM, propriétaire, hors de ce dépôt.
|
|
|
|
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.2'
|
|
MAJEURE = VERSION.split('.')[0]
|
|
NOM = 'etm-powersync-ref-optimizer'
|
|
# Nom annoncé au handshake : `powersync-optimizer`, l'exemple GPL de référence. `etm-optimizer`
|
|
# est le nom de l'optimiseur ETM, propriétaire, hors de ce dépôt.
|
|
NOM_ANNONCE = 'powersync-optimizer'
|
|
LIGNE_MAX = 1024 * 1024 # E-T2-1, §2 : 1 Mio par ligne, '\n' non compté, dans les deux sens
|
|
CODE_TOKEN_REFUSE = -32001 # E-T2-6, §3
|
|
CODE_ANALYSE = -32700 # JSON-RPC 2.0 ; codes 1.2 : OPTIMIZER_PROTOCOL.md §7
|
|
CODE_REQUETE_INVALIDE = -32600
|
|
CODE_METHODE_INCONNUE = -32601
|
|
CODE_PARAMS_INVALIDES = -32602
|
|
TROP_LONGUE = object() # lire_ligne() : ligne de plus de LIGNE_MAX octets, déjà jetée
|
|
|
|
MODES = ['normal', 'abstain', 'slow', 'invalid', 'expired', 'wrong-major', 'drop', 'unknown-load',
|
|
'forbidden-kind', 'missing-load', 'echo-token', 'protocol-1.0', 'deviate', 'wrong-type']
|
|
|
|
# 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 problemes_contexte(ctx):
|
|
"""Ce que cet optimiseur ne peut pas lire dans un contexte GetPlan : [{path, problem}], jamais
|
|
la valeur reçue (§7, 1.2). Seul `timestamp` est exigé — tout le reste se lit avec un défaut."""
|
|
if instant(ctx.get('timestamp')) is None:
|
|
return [{'path': '/timestamp', 'problem': 'absent, illisible ou sans décalage horaire'}]
|
|
return []
|
|
|
|
|
|
def plan(ctx, mode, horizon_s, token):
|
|
"""La réponse à un GetPlan (`result`) ; le contexte a passé problemes_contexte()."""
|
|
t0 = time.monotonic()
|
|
maintenant = instant(ctx.get('timestamp'))
|
|
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 == 'wrong-type':
|
|
resultat['planId'] = 42
|
|
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
|
|
|
|
|
|
VERROU_CAPTURE = threading.Lock()
|
|
|
|
|
|
def capturer(cfg, message):
|
|
"""--capture : une ligne par message. Le token reçu au handshake, et toute recopie du token
|
|
configuré (mode echo-token), deviennent autant d'astérisques : forme gardée, valeur jamais écrite."""
|
|
if not cfg.capture:
|
|
return
|
|
if isinstance(message, dict) and isinstance(message.get('params'), dict) \
|
|
and isinstance(message['params'].get('token'), str):
|
|
message = dict(message, params=dict(message['params'], token='*' * len(message['params']['token'])))
|
|
ligne = json.dumps(message, separators=(',', ':'), ensure_ascii=False)
|
|
token = lire_token(cfg.token_file)
|
|
if token:
|
|
for forme in (token, json.dumps(token, ensure_ascii=False)[1:-1]):
|
|
ligne = ligne.replace(forme, '*' * len(forme))
|
|
with VERROU_CAPTURE, open(cfg.capture, 'a', encoding='utf-8') as f:
|
|
f.write(ligne + '\n')
|
|
|
|
|
|
class Connexion(socketserver.StreamRequestHandler):
|
|
"""Une connexion = un fil ; requêtes traitées dans l'ordre d'arrivée."""
|
|
|
|
def envoyer(self, message):
|
|
capturer(self.server.cfg, message)
|
|
ligne = json.dumps(message, separators=(',', ':'), ensure_ascii=False).encode() + b'\n'
|
|
self.wfile.write(ligne)
|
|
self.wfile.flush()
|
|
|
|
def erreur(self, ident, code, message, data=None):
|
|
e = {'code': code, 'message': message}
|
|
if data is not None:
|
|
e['data'] = data
|
|
self.envoyer({'jsonrpc': '2.0', 'id': ident, 'error': e})
|
|
|
|
def lire_ligne(self):
|
|
"""Une ligne sans son '\n' ; None en fin de flux ; TROP_LONGUE pour une ligne de plus de
|
|
LIGNE_MAX octets, lue jusqu'à sa fin par morceaux, sans jamais être gardée entière."""
|
|
morceau = self.rfile.readline(LIGNE_MAX + 1)
|
|
if not morceau:
|
|
return None
|
|
if morceau.endswith(b'\n'):
|
|
return morceau[:-1]
|
|
if len(morceau) <= LIGNE_MAX:
|
|
return morceau # dernière ligne, sans '\n', avant la fin du flux
|
|
while True:
|
|
morceau = self.rfile.readline(LIGNE_MAX + 1)
|
|
if not morceau or morceau.endswith(b'\n'):
|
|
return TROP_LONGUE
|
|
|
|
def handle(self):
|
|
cfg = self.server.cfg
|
|
pret = False
|
|
token = None
|
|
while True:
|
|
try:
|
|
ligne = self.lire_ligne()
|
|
except (OSError, ValueError):
|
|
return
|
|
if ligne is None:
|
|
return
|
|
if ligne is TROP_LONGUE:
|
|
journal('ligne trop longue écartée')
|
|
self.erreur(None, CODE_REQUETE_INVALIDE, 'ligne trop longue')
|
|
continue
|
|
if not ligne.strip():
|
|
continue
|
|
try:
|
|
message = json.loads(ligne)
|
|
except ValueError:
|
|
journal('ligne non JSON écartée')
|
|
self.erreur(None, CODE_ANALYSE, "erreur d'analyse")
|
|
continue
|
|
capturer(cfg, message)
|
|
if isinstance(message, list):
|
|
journal('lot JSON-RPC écarté : un message par ligne')
|
|
self.erreur(None, CODE_REQUETE_INVALIDE, 'lot interdit')
|
|
continue
|
|
if not isinstance(message, dict):
|
|
self.erreur(None, CODE_REQUETE_INVALIDE, 'requête invalide')
|
|
continue
|
|
if 'method' not in message:
|
|
continue # une réponse : le moteur n'en envoie pas, rien à répondre
|
|
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.erreur(ident, CODE_TOKEN_REFUSE, 'token refusé')
|
|
return
|
|
version = params.get('protocolVersion')
|
|
if not isinstance(version, str) or version.split('.')[0] != MAJEURE:
|
|
journal(f'handshake refusé : majeure incompatible ({str(version)[:32]})')
|
|
self.erreur(ident, CODE_PARAMS_INVALIDES, 'majeure incompatible')
|
|
return
|
|
token = attendu
|
|
pret = True
|
|
reponse = {'protocolVersion': VERSION, 'name': NOM_ANNONCE, '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.erreur(ident, CODE_REQUETE_INVALIDE, 'handshake requis')
|
|
continue
|
|
problemes = problemes_contexte(params)
|
|
if problemes:
|
|
journal(f'contexte invalide : {", ".join(p["path"] for p in problemes)}')
|
|
self.erreur(ident, CODE_PARAMS_INVALIDES, 'contexte invalide', {'problems': problemes})
|
|
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.erreur(ident, CODE_METHODE_INCONNUE, '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.2, 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('--capture', help='écrit chaque message reçu et envoyé dans ce fichier (NDJSON)')
|
|
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())
|