vojacekd/ems
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(modbus): reg 15 re-asert kazdy tick + per-charger failsafe (BUG1)

Browse Source 
Zivy incident home-01 (TeltoCharge .16): od ~22:45 UTC 12.6. nevznikl zadny
telto journal radek (ani failed), auto jelo failsafe 8 A misto planovanych 0 A.

Root cause: reg 15 (amps) byl write-on-change proti journalu
(fn_modbus_device_state_map). Jakmile mel reg 15 radek "0 verified" a plan
dal chtel 0, NIKDY nevznikl novy prikaz -- a TeltoCharge si po vypadku
komunikace sam prepsal reg 15 na failsafe (reg 20) BEZ journal radku. Verify
cte zpet jen 'written' radky, takze tichy drift 0 -> 8 A nikdo nevidel ani
neopravil.

- reg 15 (amps to use) se zapisuje VZDY (re-asert) -- volatilni ridici
  registr, ne EEPROM; drzi verify jobu cerstvy written radek -> drift se
  zachyti a hned opravi. _split_amps_and_watchdog odděluje 15 od 19/20.
- reg 19/20 (watchdog config, EEPROM) zustavaji write-on-change.
- per-charger failsafe/timeout: asset_ev_charger.watchdog_failsafe_a /
  watchdog_comm_timeout_s (V106; default 8 A / 300 s). "Zakaz nabijeni" =
  reg 15 = 0 (protokol rev 0.5 nema samostatny enable registr).
- testy test_ev_write_on_change.py; docs teltocharge + journal + data-model.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dusan Vojacek
2026-06-13 22:03:11 +02:00
parent 03b7396676
commit 54288ee2fd
6 changed files with 229 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
backend/tests/test_ev_write_on_change.py
View File

@@ -1,11 +1,13 @@
"""Write-on-change pro TeltoCharge: zápis JEN při skutečné změně hodnoty.
"""TeltoCharge zápis: reg 15 (amps) VŽDY, watchdog 19/20 write-on-change.
Regrese na opakované zápisy do wallboxu (EEPROM wear): export tick běží
~8x/hod (control_export :14,:29,:44,:59 + rolling replan */15 s exportem),
ale reg 15/19/20 se smí zapsat jen při změně proti poslednímu known stavu
zařízení (fn_modbus_device_state_map: nejnovější written/verified řádek
journalu per registr). Watchdog (19/20) se zapíše jednou po startu / po
výpadku; sytí ho i FC3 čtení telemetrie (60 s), periodické zápisy netřeba.
Export tick běží ~8x/hod (control_export :14,:29,:44,:59 + rolling replan
*/15 s exportem). **Reg 15 (amps to use) se zapisuje VŽDY** — TeltoCharge ho
po výpadku komunikace sám přepíše na failsafe (reg 20) bez journal řádku, a
kdyby byl write-on-change, EMS by tichý drift 0 → 8 A nikdy nezahlédlo
(verify čte zpět jen `written`). **Reg 19/20 (watchdog config, EEPROM wear)
zůstávají write-on-change** proti fn_modbus_device_state_map (nejnovější
written/verified řádek per registr): zapíší se jednou po startu / po výpadku;
sytí je i FC3 čtení telemetrie (60 s), periodické zápisy netřeba.
"""
import unittest
@@ -20,6 +22,7 @@ from services.control.outputs import (
TELTO_REG_FAILSAFE_CURRENT_A,
TELTO_WATCHDOG_FAILSAFE_A,
TELTO_WATCHDOG_TIMEOUT_S,
_split_amps_and_watchdog,
_telto_setpoint_registers,
write_ev_arrival_hold,
write_ev_setpoints,
@@ -59,40 +62,48 @@ class TeltoSetpointRegistersTests(unittest.TestCase):
self.assertEqual(_telto_setpoint_registers(6)[0][2], 6)
self.assertEqual(_telto_setpoint_registers(40)[0][2], 32)
def test_per_charger_failsafe_and_timeout(self) -> None:
regs = _telto_setpoint_registers(0, comm_timeout_s=120, failsafe_a=6)
self.assertEqual([(r, v) for r, _, v in regs], [(15, 0), (19, 120), (20, 6)])
def test_failsafe_clamped_to_0_32(self) -> None:
self.assertEqual(_telto_setpoint_registers(0, failsafe_a=99)[2][2], 32)
self.assertEqual(_telto_setpoint_registers(0, failsafe_a=-5)[2][2], 0)
def test_split_separates_amps_from_watchdog(self) -> None:
amps, watchdog = _split_amps_and_watchdog(_telto_setpoint_registers(0))
self.assertEqual([r for r, _, _ in amps], [15])
self.assertEqual([r for r, _, _ in watchdog], [19, 20])
class DropAgainstDeviceStateTests(unittest.TestCase):
def test_steady_state_drops_everything(self) -> None:
def test_watchdog_steady_state_drops_19_20(self) -> None:
_, watchdog = _split_amps_and_watchdog(_telto_setpoint_registers(0))
out, skipped = _drop_registers_matching_last_verified(
_telto_setpoint_registers(0), _STEADY_STATE_0A
watchdog, _STEADY_STATE_0A
)
self.assertEqual(out, [])
self.assertEqual(skipped, [15, 19, 20])
def test_amps_change_writes_only_reg_15(self) -> None:
out, skipped = _drop_registers_matching_last_verified(
_telto_setpoint_registers(16), _STEADY_STATE_0A
)
self.assertEqual([(r, v) for r, _, v in out], [(15, 16)])
self.assertEqual(skipped, [19, 20])
def test_empty_state_after_outage_writes_full_triple(self) -> None:
out, skipped = _drop_registers_matching_last_verified(
_telto_setpoint_registers(0), {}
)
self.assertEqual([r for r, _, _ in out], [15, 19, 20])
def test_empty_state_after_outage_keeps_19_20(self) -> None:
_, watchdog = _split_amps_and_watchdog(_telto_setpoint_registers(0))
out, skipped = _drop_registers_matching_last_verified(watchdog, {})
self.assertEqual([r for r, _, _ in out], [19, 20])
self.assertEqual(skipped, [])
class _FakeDB:
"""Jen řádky chargeru; journal funkce se patchují v modbus_journal."""
def __init__(self) -> None:
def __init__(self, failsafe_a: int = 8, comm_timeout_s: int = 300) -> None:
self.row = {
"asset_id": 7,
"code": "ev-charger-1",
"host": "172.16.1.16",
"port": 502,
"unit_id": 1,
"watchdog_failsafe_a": failsafe_a,
"watchdog_comm_timeout_s": comm_timeout_s,
}
async def fetch(self, query: str, *args: object) -> list[dict]:
@@ -105,9 +116,9 @@ class _FakeDB:
raise AssertionError(f"unexpected fetchval: {query}")
class WriteEvSetpointsOnChangeTests(unittest.IsolatedAsyncioTestCase):
class WriteEvSetpointsTests(unittest.IsolatedAsyncioTestCase):
async def _run(
self, device_state: dict[int, int], ev1_a: int
self, device_state: dict[int, int], ev1_a: int, db: _FakeDB | None = None
) -> tuple[AsyncMock, AsyncMock]:
create = AsyncMock(return_value=[1, 2, 3])
execute = AsyncMock(return_value=True)
@@ -120,13 +131,17 @@ class WriteEvSetpointsOnChangeTests(unittest.IsolatedAsyncioTestCase):
patch.object(journal, "create_modbus_commands", create),
patch.object(journal, "execute_modbus_commands", execute),
):
await write_ev_setpoints(1, _setpoints(ev1_a), _FakeDB()) # type: ignore[arg-type]
await write_ev_setpoints(1, _setpoints(ev1_a), db or _FakeDB()) # type: ignore[arg-type]
return create, execute
async def test_steady_state_no_write_at_all(self) -> None:
async def test_steady_state_still_reasserts_reg_15(self) -> None:
# Reg 15 se zapisuje VŽDY (re-asert proti tichému failsafe driftu),
# i když je device-state mapa shodná. Watchdog 19/20 se přeskočí.
create, execute = await self._run(_STEADY_STATE_0A, ev1_a=0)
create.assert_not_awaited()
execute.assert_not_awaited()
create.assert_awaited_once()
registers = create.await_args.args[8]
self.assertEqual([(r, v) for r, _, v in registers], [(15, 0)])
execute.assert_awaited_once()
async def test_plan_change_writes_only_amps(self) -> None:
create, execute = await self._run(_STEADY_STATE_0A, ev1_a=16)
@@ -135,14 +150,24 @@ class WriteEvSetpointsOnChangeTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual([(r, v) for r, _, v in registers], [(15, 16)])
execute.assert_awaited_once()
async def test_after_outage_writes_full_triple(self) -> None:
async def test_after_outage_writes_amps_then_watchdog(self) -> None:
create, execute = await self._run({}, ev1_a=0)
registers = create.await_args.args[8]
self.assertEqual([r for r, _, _ in registers], [15, 19, 20])
execute.assert_awaited_once()
async def test_per_charger_failsafe_from_db(self) -> None:
# Failsafe 6 A z DB → po výpadku se zapíše reg 20 = 6 (prázdná mapa).
create, _ = await self._run(
{}, ev1_a=0, db=_FakeDB(failsafe_a=6, comm_timeout_s=120)
)
registers = create.await_args.args[8]
self.assertEqual(
[(r, v) for r, _, v in registers], [(15, 0), (19, 120), (20, 6)]
)
class WriteEvArrivalHoldOnChangeTests(unittest.IsolatedAsyncioTestCase):
class WriteEvArrivalHoldTests(unittest.IsolatedAsyncioTestCase):
async def _run(
self, device_state: dict[int, int]
) -> tuple[bool, AsyncMock, AsyncMock]:
@@ -160,13 +185,15 @@ class WriteEvArrivalHoldOnChangeTests(unittest.IsolatedAsyncioTestCase):
ok = await write_ev_arrival_hold(1, "ev-charger-1", _FakeDB()) # type: ignore[arg-type]
return ok, create, execute
async def test_hold_skips_when_device_already_at_zero(self) -> None:
async def test_hold_always_writes_reg_15_even_if_device_at_zero(self) -> None:
# Tvrdé zastavení po píchnutí kabelu — reg 15 = 0 se zapíše VŽDY.
ok, create, execute = await self._run(_STEADY_STATE_0A)
self.assertTrue(ok)
create.assert_not_awaited()
execute.assert_not_awaited()
registers = create.await_args.args[8]
self.assertEqual([(r, v) for r, _, v in registers], [(15, 0)])
execute.assert_awaited_once()
async def test_hold_writes_only_amps_when_watchdog_already_set(self) -> None:
async def test_hold_writes_amps_and_watchdog_when_device_drifted(self) -> None:
ok, create, execute = await self._run(
{
TELTO_REG_AMPS_TO_USE: 16,
@@ -176,6 +203,7 @@ class WriteEvArrivalHoldOnChangeTests(unittest.IsolatedAsyncioTestCase):
)
self.assertTrue(ok)
registers = create.await_args.args[8]
# Reg 15 = 0 zapsán (i když device hlásí 16); 19/20 shodné → skip.
self.assertEqual([(r, v) for r, _, v in registers], [(15, 0)])
execute.assert_awaited_once()