Merge branch 'fix/ev-teltocharge-reg15-and-session-visibility' into dev

# Conflicts:
#	docs/planning-changelog.md

backend/tests/test_ev_session_parse.py

@@ -0,0 +1,66 @@
+"""Parser EV session z fn_planning_site_context (_ev_session_from_json).
+
+Bug 2026-06-13: session BEZ deadline (auto nad targetem / bez cíle) se v
+parseru zahazovala (None), takže plánovač neviděl zátěž auta ani oportunismus.
+Oprava: session bez deadline zůstává objektem s energy_needed_wh=0 a headroom.
+"""
+
+import unittest
+
+from services.planning.db_io import _ev_session_from_json
+
+
+class EvSessionParseTests(unittest.TestCase):
+    def test_none_and_empty_return_none(self) -> None:
+        self.assertIsNone(_ev_session_from_json(None))
+        self.assertIsNone(_ev_session_from_json([]))
+        self.assertIsNone(_ev_session_from_json(123))
+
+    def test_session_without_deadline_kept_for_opportunism(self) -> None:
+        sess = _ev_session_from_json(
+            {
+                "target_deadline": None,
+                "energy_needed_wh": 0,
+                "headroom_wh": 18000.0,
+                "opportunistic_value_czk_kwh": 1.0,
+            }
+        )
+        self.assertIsNotNone(sess)
+        assert sess is not None
+        self.assertIsNone(sess.target_deadline)
+        self.assertEqual(sess.energy_needed_wh, 0.0)
+        self.assertEqual(sess.headroom_wh, 18000.0)
+        self.assertEqual(sess.opportunistic_value_czk_kwh, 1.0)
+
+    def test_session_with_deadline_and_need(self) -> None:
+        sess = _ev_session_from_json(
+            {
+                "target_deadline": "2026-06-14T05:00:00+00:00",
+                "energy_needed_wh": 12000.0,
+                "headroom_wh": 6000.0,
+                "opportunistic_value_czk_kwh": 1.0,
+            }
+        )
+        assert sess is not None
+        self.assertIsNotNone(sess.target_deadline)
+        self.assertEqual(sess.energy_needed_wh, 12000.0)
+
+    def test_missing_needed_defaults_zero(self) -> None:
+        sess = _ev_session_from_json(
+            {"target_deadline": None, "headroom_wh": 1000.0}
+        )
+        assert sess is not None
+        self.assertEqual(sess.energy_needed_wh, 0.0)
+        self.assertEqual(sess.opportunistic_value_czk_kwh, 0.0)
+
+    def test_json_string_payload(self) -> None:
+        sess = _ev_session_from_json(
+            '{"target_deadline": null, "energy_needed_wh": 0, '
+            '"headroom_wh": 5000, "opportunistic_value_czk_kwh": 1.0}'
+        )
+        assert sess is not None
+        self.assertEqual(sess.headroom_wh, 5000.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_ev_write_on_change.py

@@ -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()