"""Export plánovaných setpointů na Modbus (Deye, EV, TČ) a HTTP do Loxone.""" from __future__ import annotations import asyncio import logging import os from collections import defaultdict from dataclasses import dataclass from typing import Any from datetime import date, datetime, timedelta, timezone from zoneinfo import ZoneInfo import asyncpg import httpx from app.config import get_settings from services.modbus_client import get_modbus_client logger = logging.getLogger(__name__) PRAGUE_TZ = ZoneInfo("Europe/Prague") # Hodiny Deye 62–64: po zápisu sekundy na zařízení dál běží → verify musí být toleranční. DEYE_CLOCK_VERIFY_MAX_DELTA_SEC = 120 # Řidší zápis: bez zápisu, pokud čas na invertoru neodbočí od Prahy víc než o tolik sekund… DEYE_CLOCK_DRIFT_OK_SEC = 60 # …a zároveň neuplynul tento interval od posledního syncu / potvrzení driftu. DEYE_CLOCK_RESYNC_INTERVAL_HOURS = 24 # Deye LV baterie: převod výkon → proud pro registry 108/109 (viz docs/04-modules/modbus-registers.md) BATT_VOLTAGE_V = 51.2 # Reg 178 – pevné hodnoty (bit4–5); bez read-modify-write (kolize s Loxone / transaction ID) REG178_SELL = 0b00100000 # 32, grid peak shaving disable REG178_PASSIVE = 0b00110000 # 48, grid peak shaving enable (PASSIVE i CHARGE) # Neaktivní TOU bloky (3–6): „konec dne“ — Deye často 23:59 (2359) neuloží a vrátí např. 2355, # verify pak hlásí mismatch. 23:55 je na zařízeních stabilní (viz HHMM jako desítkové číslo). DEYE_TOU_INACTIVE_HHMM = 2355 # Registry TOU řádků 3–6 (slot index 2…5): 150–153, 156–159, … — pro detekci skutečného zápisu po filtru „unchanged“. _DEYE_INACTIVE_TOU_REGISTERS: frozenset[int] = frozenset( [ 150, 151, 152, 153, 156, 157, 158, 159, 168, 169, 170, 171, 174, 175, 176, 177, ] ) DEYE_REGISTER_NAMES: dict[int, str] = { 108: "max_charge_a (max nabíjecí proud baterie)", 109: "max_discharge_a (max vybíjecí proud baterie)", 141: "energy_mode (0, EMS nemění)", 142: "limit_control (0=selling first, 1=zero export built-in CT)", 143: "export_limit_w (max export do sítě)", 178: "grid_peak_shaving_switch (SELL=32 bit4-5=10, PASSIVE/CHARGE=48 bit4-5=11)", 148: "time_point_1_time", 149: "time_point_2_time", 154: "time_point_1_power_w", 155: "time_point_2_power_w", 166: "time_point_1_soc_min_pct", 167: "time_point_2_soc_min_pct", 172: "time_point_1_grid_charge", 173: "time_point_2_grid_charge", 62: "system_time_year_month", 63: "system_time_day_hour", 64: "system_time_min_sec", } for _tp_i in range(6): _n = _tp_i + 1 DEYE_REGISTER_NAMES.setdefault(148 + _tp_i, f"time_point_{_n}_time") DEYE_REGISTER_NAMES.setdefault(154 + _tp_i, f"time_point_{_n}_power_w") DEYE_REGISTER_NAMES.setdefault(166 + _tp_i, f"time_point_{_n}_soc_min_pct") DEYE_REGISTER_NAMES.setdefault(172 + _tp_i, f"time_point_{_n}_grid_charge") def watts_to_amps(power_w: int | None, phases: int = 3, voltage: int = 230) -> int: if not power_w or power_w <= 0: return 0 return min(32, max(0, int(power_w / (phases * voltage)))) def battery_watts_to_amps(power_w: int, max_amps: int) -> int: """Proud z |výkonu| baterie; max_amps výhradně z DB (_load_inverter_config).""" return min(max(0, max_amps), max(0, round(abs(power_w) / BATT_VOLTAGE_V))) def current_slot_hhmm() -> int: """Začátek probíhajícího 15min slotu v Europe/Prague, formát HHMM (např. 1415).""" now = datetime.now(ZoneInfo("Europe/Prague")) slot_min = (now.minute // 15) * 15 return now.hour * 100 + slot_min def next_slot_hhmm() -> int: """Začátek příštího 15min slotu v Europe/Prague, formát HHMM (např. 1430).""" now = datetime.now(ZoneInfo("Europe/Prague")) minutes = now.minute slot_minutes = ((minutes // 15) + 1) * 15 if slot_minutes >= 60: next_hour = (now.hour + 1) % 24 next_min = 0 else: next_hour = now.hour next_min = slot_minutes return next_hour * 100 + next_min @dataclass class InverterConfig: id: int code: str host: str port: int unit_id: int max_export_power_w: int | None max_import_power_w: int | None no_export: bool max_battery_charge_w: int | None max_battery_discharge_w: int | None min_soc_percent: int | None reserve_soc_percent: int | None max_soc_percent: int | None usable_capacity_wh: int | None max_charge_a: int max_discharge_a: int deye_last_system_time_sync_minute: datetime | None = None deye_last_system_time_sync_at: datetime | None = None deye_last_tou_inactive_write_prague_date: date | None = None deye_tou_inactive_signature: str | None = None def _prague_minute_start_utc() -> datetime: """UTC okamžik odpovídající začátku aktuální kalendářní minuty v Europe/Prague.""" p = datetime.now(PRAGUE_TZ).replace(second=0, microsecond=0) return p.astimezone(timezone.utc) def _deye_registers_to_prague_datetime(r62: int, r63: int, r64: int) -> datetime | None: """Dekódování reg 62–64 (Deye system time v Europe/Prague).""" try: year = (int(r62) >> 8) + 2000 month = int(r62) & 0xFF day = int(r63) >> 8 hour = int(r63) & 0xFF minute = int(r64) >> 8 second = int(r64) & 0xFF if not (1 <= month <= 12 and 1 <= day <= 31 and 0 <= hour <= 23): return None if not (0 <= minute <= 59 and 0 <= second <= 59): return None return datetime(year, month, day, hour, minute, second, tzinfo=PRAGUE_TZ) except (ValueError, OverflowError): return None def _deye_clock_registers_verify_match( w62: int, w63: int, w64: int, a62: int, a63: int, a64: int, ) -> bool: w_dt = _deye_registers_to_prague_datetime(w62, w63, w64) a_dt = _deye_registers_to_prague_datetime(a62, a63, a64) if w_dt is None or a_dt is None: return False return abs((a_dt - w_dt).total_seconds()) <= DEYE_CLOCK_VERIFY_MAX_DELTA_SEC def _deye_should_skip_time_sync_after_read( inv: InverterConfig, r62: int, r63: int, r64: int, ) -> bool: """ True = nezařazovat zápis 62–64: drift je malý a od posledního úspěšného ověření času (status verified v journalu 62–64) neuplynul 24h — sloupec deye_last_system_time_sync_at se doplňuje jen po tolerančním ověření v _verify_deye_clock_command_run. """ dev = _deye_registers_to_prague_datetime(r62, r63, r64) if dev is None: return False wall = datetime.now(PRAGUE_TZ) drift = abs((wall - dev).total_seconds()) if drift > DEYE_CLOCK_DRIFT_OK_SEC: return False last_write = inv.deye_last_system_time_sync_at if last_write is None: return False if last_write.tzinfo is None: last_write = last_write.replace(tzinfo=timezone.utc) else: last_write = last_write.astimezone(timezone.utc) age = datetime.now(timezone.utc) - last_write if age >= timedelta(hours=DEYE_CLOCK_RESYNC_INTERVAL_HOURS): return False return True def _is_deye_contiguous_clock_run(run: list[asyncpg.Record]) -> bool: if len(run) != 3: return False regs = sorted(int(c["register"]) for c in run) return regs == [62, 63, 64] async def _fetch_last_verified_inverter_registers( site_id: int, inverter_asset_id: int, db: asyncpg.Connection ) -> dict[int, int]: """ Poslední hodnota na zařízení podle journalu (jen status verified). Slouží k přeskočení duplicitního zápisu stejné hodnoty. """ rows = await db.fetch( """ SELECT DISTINCT ON (register) register, value_verified FROM ems.modbus_command WHERE site_id = $1 AND asset_type = 'inverter' AND asset_id = $2 AND status = 'verified' AND value_verified IS NOT NULL ORDER BY register, verified_at DESC NULLS LAST, id DESC """, site_id, inverter_asset_id, ) return {int(r["register"]): int(r["value_verified"]) for r in rows} def _drop_registers_matching_last_verified( registers: list[tuple[int, str, int]], last_verified: dict[int, int], ) -> tuple[list[tuple[int, str, int]], list[int]]: """Vynechá položky s hodnotou shodnou s posledním ověřeným stavem; vrátí (nový seznam, vynechané reg).""" out: list[tuple[int, str, int]] = [] skipped: list[int] = [] for reg, meta, val in registers: lv = last_verified.get(int(reg)) if lv is not None and lv == int(val): skipped.append(int(reg)) continue out.append((reg, meta, val)) return out, skipped @dataclass class ControlSetpoints: battery_w: int | None grid_export_limit: int ev1_current_a: int ev2_current_a: int heat_pump_enable: bool grid_setpoint_w: int ev1_power_w: int ev2_power_w: int target_soc_pct: int | None = None #: True = reg 108/109 na 0 (PRESERVE – Deye baterii nepoužívá) lock_battery: bool = False @dataclass class OperatingModeInfo: mode_code: str battery_mode: str grid_mode: str ev_enabled: bool heat_pump_enabled_def: bool loxone_mode_value: int async def create_modbus_commands( site_id: int, planning_run_id: int | None, asset_type: str, asset_id: int, asset_code: str, host: str, port: int, unit_id: int, registers: list[tuple[int, str, int]], db: asyncpg.Connection, deye_physical_mode: str | None = None, ) -> list[int]: """ Vytvoří záznamy v modbus_command pro sadu zápisů. Vrátí list command IDs. Pro Deye se jméno registru bere z DEYE_REGISTER_NAMES (prostřední položka tuplu se ignoruje). """ ids: list[int] = [] for reg, _ignored_name, val in registers: register_name = DEYE_REGISTER_NAMES.get(reg, f"reg_{reg}") cmd_id = await db.fetchval( """ INSERT INTO ems.modbus_command (site_id, asset_type, asset_id, asset_code, device_host, device_port, device_unit_id, register, register_name, value_to_write, planning_run_id, status, deye_physical_mode) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,'pending',$12) RETURNING id """, site_id, asset_type, asset_id, asset_code, host, port, unit_id, reg, register_name, val, planning_run_id, deye_physical_mode, ) if cmd_id is not None: ids.append(int(cmd_id)) return ids def _modbus_command_contiguous_runs(cmds: list[asyncpg.Record]) -> list[list[asyncpg.Record]]: """Seřadí podle adresy registru a rozdělí na souvislé úseky pro FC 0x10 / FC 3.""" if not cmds: return [] sorted_cmds = sorted(cmds, key=lambda c: int(c["register"])) runs: list[list[asyncpg.Record]] = [] cur: list[asyncpg.Record] = [sorted_cmds[0]] for c in sorted_cmds[1:]: if int(c["register"]) == int(cur[-1]["register"]) + 1: cur.append(c) else: runs.append(cur) cur = [c] runs.append(cur) return runs async def execute_modbus_commands( command_ids: list[int], db: asyncpg.Connection, ) -> bool: """ Zapíše příkazy z modbus_command do zařízení (FC 0x10 po souvislých blocích). Aktualizuje status na 'written' nebo 'failed'. Vrátí True pokud všechny příkazy uspěly. """ MAX_RETRIES = 3 RETRY_DELAY = 0.5 rows: list[asyncpg.Record] = [] for cmd_id in command_ids: cmd = await db.fetchrow( "SELECT * FROM ems.modbus_command WHERE id=$1", cmd_id ) if cmd is not None: rows.append(cmd) if not rows: return True by_gw: dict[tuple[str, int, int], list[asyncpg.Record]] = defaultdict(list) for cmd in rows: by_gw[ (cmd["device_host"], int(cmd["device_port"]), int(cmd["device_unit_id"])) ].append(cmd) all_ok = True for (host, port, unit), group in by_gw.items(): client = await get_modbus_client(host, port) for run in _modbus_command_contiguous_runs(group): start_reg = int(run[0]["register"]) values = [int(c["value_to_write"]) for c in run] ids_run = [int(c["id"]) for c in run] for attempt in range(MAX_RETRIES): try: await client.write_registers(start_reg, values, unit) for cmd, val in zip(run, values): cid = int(cmd["id"]) await db.execute( """ UPDATE ems.modbus_command SET status='written', value_written=$1, written_at=now(), attempt_count=attempt_count+1, error_msg=NULL WHERE id=$2 """, val, cid, ) logger.info( "[cmd %s] %s 0x%04X=%s OK batch@%s (attempt %s)", cid, cmd["asset_code"], int(cmd["register"]), val, start_reg, attempt + 1, ) break except Exception as e: if attempt < MAX_RETRIES - 1: logger.warning( "Modbus batch write 0x%04X count=%s attempt %s failed: %s, retrying...", start_reg, len(values), attempt + 1, e, ) await asyncio.sleep(RETRY_DELAY) await client.force_disconnect() else: for cmd in run: await db.execute( """ UPDATE ems.modbus_command SET status='failed', error_msg=$1, attempt_count=attempt_count+1 WHERE id=$2 """, str(e), int(cmd["id"]), ) logger.error( "Modbus batch 0x%04X count=%s all %s attempts failed: %s", start_reg, len(values), MAX_RETRIES, e, ) all_ok = False return all_ok async def _switch_to_self_sustain(site_id: int, db: asyncpg.Connection, reason: str) -> None: """Přepne lokalitu na SELF_SUSTAIN, zaloguje důvod a při změně pošle Discord.""" from services.notification_service import run_fn_set_mode_with_discord await run_fn_set_mode_with_discord( db, site_id, "SELF_SUSTAIN", "system:mismatch", None, reason, ) logger.critical("Site %s switched to SELF_SUSTAIN: %s", site_id, reason) async def _verify_deye_clock_command_run( run: list[asyncpg.Record], values: list[int], db: asyncpg.Connection, site_id: int, ) -> bool: """ Ověření souvislého bloku 62–64: porovnání času z trojice registrů s tolerancí (sekundy na Deye běží). Při mismatch retry všech tří řádků journalu společně. """ from services.notification_service import ( notify_modbus_clock_verify_exhausted, notify_modbus_mismatch, ) run_s = sorted(run, key=lambda c: int(c["register"])) w62 = int(run_s[0]["value_to_write"]) w63 = int(run_s[1]["value_to_write"]) w64 = int(run_s[2]["value_to_write"]) a62, a63, a64 = (int(values[0]), int(values[1]), int(values[2])) clock_ok = _deye_clock_registers_verify_match(w62, w63, w64, a62, a63, a64) for cmd, actual in zip(run_s, values): cid = int(cmd["id"]) await db.execute( """ UPDATE ems.modbus_command SET value_verified=$1::int, verified_at=now(), status=CASE WHEN $2::boolean THEN 'verified' ELSE 'mismatch' END WHERE id=$3::int """, int(actual), clock_ok, cid, ) if clock_ok: inv_asset_id = int(run_s[0]["asset_id"]) await db.execute( """ UPDATE ems.asset_inverter SET deye_last_system_time_sync_minute = $1, deye_last_system_time_sync_at = now() WHERE id = $2 """, _prague_minute_start_utc(), inv_asset_id, ) for cmd, actual in zip(run_s, values): logger.info( "[cmd %s] verified OK (clock tolerant): %s 0x%04X=%s", int(cmd["id"]), cmd["asset_code"], int(cmd["register"]), int(actual), ) return True cmd0 = run_s[0] logger.error( "[cmd clock] MISMATCH %s 62–64: written=(%s,%s,%s) actual=(%s,%s,%s)", cmd0["asset_code"], w62, w63, w64, a62, a63, a64, ) attempts = 0 for cmd in run_s: row_ac = await db.fetchrow( "SELECT attempt_count FROM ems.modbus_command WHERE id=$1", int(cmd["id"]) ) ac = int(row_ac["attempt_count"] or 0) if row_ac else 0 attempts = max(attempts, ac) await notify_modbus_mismatch( str(cmd0["asset_code"]), 62, "system_time_62_64", w62, a62, attempts, ) ids_ordered = [int(c["id"]) for c in run_s] if attempts < 3: for cid in ids_ordered: await db.execute( "UPDATE ems.modbus_command SET status='retrying' WHERE id=$1", cid, ) await execute_modbus_commands(ids_ordered, db) await verify_modbus_commands(ids_ordered, db, site_id) else: logger.critical( "[cmd clock] 3 failed verify attempts (62–64); režim se nemění automaticky" ) site = await db.fetchrow("SELECT code FROM ems.site WHERE id=$1", site_id) await notify_modbus_clock_verify_exhausted( site["code"] if site else str(site_id), str(cmd0["asset_code"]), (w62, w63, w64), (a62, a63, a64), ) return False async def verify_modbus_commands( command_ids: list[int], db: asyncpg.Connection, site_id: int, ) -> bool: """ Přečte registry zpět (FC 3 po souvislých blocích) a porovná s value_to_write. Při mismatch: retry → SELF_SUSTAIN + Discord. """ from services.notification_service import notify_modbus_mismatch async def _apply_verify_result(cmd: asyncpg.Record, actual_i: int) -> bool: """Vrátí True při shodě, False při mismatch (a obslouží retry / SELF_SUSTAIN).""" cmd_id = int(cmd["id"]) expected_i = int(cmd["value_to_write"]) await db.execute( """ UPDATE ems.modbus_command SET value_verified=$1::int, verified_at=now(), status=CASE WHEN $1::int = $2::int THEN 'verified' ELSE 'mismatch' END WHERE id=$3::int """, actual_i, expected_i, cmd_id, ) if actual_i != expected_i: logger.error( "[cmd %s] MISMATCH %s 0x%04X: expected=%s actual=%s", cmd_id, cmd["asset_code"], int(cmd["register"]), expected_i, actual_i, ) row_ac = await db.fetchrow( "SELECT attempt_count FROM ems.modbus_command WHERE id=$1", cmd_id ) attempts = int(row_ac["attempt_count"] or 0) if row_ac else 0 await notify_modbus_mismatch( cmd["asset_code"], int(cmd["register"]), cmd["register_name"] or "", expected_i, actual_i, attempts, ) if attempts < 3: await db.execute( "UPDATE ems.modbus_command SET status='retrying' WHERE id=$1", cmd_id, ) await execute_modbus_commands([cmd_id], db) await verify_modbus_commands([cmd_id], db, site_id) else: logger.critical( "[cmd %s] 3 failed attempts, switching to SELF_SUSTAIN", cmd_id, ) await _switch_to_self_sustain( site_id, db, reason=( f"Modbus mismatch po 3 pokusech: {cmd['asset_code']} " f"reg 0x{cmd['register']:04X}" ), ) return False logger.info( "[cmd %s] verified OK: %s 0x%04X=%s", cmd_id, cmd["asset_code"], int(cmd["register"]), actual_i, ) return True cmds: list[asyncpg.Record] = [] for cmd_id in command_ids: cmd = await db.fetchrow( "SELECT * FROM ems.modbus_command WHERE id=$1", cmd_id ) if cmd is not None and cmd["status"] == "written": cmds.append(cmd) if not cmds: return True by_gw: dict[tuple[str, int, int], list[asyncpg.Record]] = defaultdict(list) for cmd in cmds: by_gw[ (cmd["device_host"], int(cmd["device_port"]), int(cmd["device_unit_id"])) ].append(cmd) all_ok = True for (host, port, unit), group in by_gw.items(): client = await get_modbus_client(host, port) for run in _modbus_command_contiguous_runs(group): start_reg = int(run[0]["register"]) n = len(run) try: values = await client.read_holding_registers(start_reg, n, unit) except Exception as e: logger.error( "verify batch read 0x%04X count=%s failed: %s", start_reg, n, e ) all_ok = False continue if len(values) != n: logger.error( "verify read 0x%04X: expected %s regs, got %s", start_reg, n, len(values), ) all_ok = False continue if _is_deye_contiguous_clock_run(run): matched = await _verify_deye_clock_command_run(run, values, db, site_id) if not matched: all_ok = False continue for cmd, actual in zip(run, values): matched = await _apply_verify_result(cmd, int(actual)) if not matched: all_ok = False return all_ok async def _fetch_operating_mode(site_id: int, db: asyncpg.Connection) -> OperatingModeInfo | None: sql = """ SELECT som.mode_code, omd.battery_mode, omd.grid_mode, omd.ev_enabled, omd.heat_pump_enabled, omd.loxone_mode_value, som.valid_until FROM ems.site_operating_mode som JOIN ems.operating_mode_def omd ON omd.code = som.mode_code WHERE som.site_id = $1 """ row = await db.fetchrow(sql, site_id) if row is None: return None vu = row["valid_until"] if vu is not None: now_utc = datetime.now(timezone.utc) if vu.tzinfo is None: vu = vu.replace(tzinfo=timezone.utc) if vu <= now_utc: exp_rows = await db.fetch("SELECT * FROM ems.fn_expire_modes()") from services.notification_service import notify_operating_mode_changed for er in exp_rows: await notify_operating_mode_changed( str(er["site_code"]), str(er["old_mode"]), str(er["new_mode"]), "system:expiry", "Automatické vypršení dočasného režimu", ) row = await db.fetchrow(sql, site_id) if row is None: return None return OperatingModeInfo( mode_code=row["mode_code"], battery_mode=row["battery_mode"], grid_mode=row["grid_mode"], ev_enabled=bool(row["ev_enabled"]), heat_pump_enabled_def=bool(row["heat_pump_enabled"]), loxone_mode_value=int(row["loxone_mode_value"]), ) async def _get_current_soc(site_id: int, db: asyncpg.Connection) -> int: soc = await db.fetchval( """ SELECT battery_soc_percent FROM ems.telemetry_inverter WHERE site_id = $1 AND battery_soc_percent IS NOT NULL ORDER BY measured_at DESC LIMIT 1 """, site_id, ) return int(soc) if soc is not None else 50 async def _load_inverter_config( site_id: int, db: asyncpg.Connection ) -> InverterConfig | None: row = await db.fetchrow( """ SELECT ai.id, ai.code, se.host, se.port, se.unit_id, sgc.max_export_power_w, sgc.max_import_power_w, sgc.no_export, ai.max_battery_charge_w, ai.max_battery_discharge_w, ab.min_soc_percent, ab.reserve_soc_percent, ab.max_soc_percent, ab.usable_capacity_wh, ai.deye_last_system_time_sync_minute, ai.deye_last_system_time_sync_at, ai.deye_last_tou_inactive_write_prague_date, ai.deye_tou_inactive_signature, LEAST( COALESCE(ab.bms_max_charge_w, ai.max_battery_charge_w), ai.max_battery_charge_w ) / 51.2 AS max_charge_a, LEAST( COALESCE(ab.bms_max_discharge_w, ai.max_battery_discharge_w), ai.max_battery_discharge_w ) / 51.2 AS max_discharge_a FROM ems.asset_inverter ai JOIN ems.site_endpoint se ON se.id = ai.endpoint_id JOIN ems.asset_battery ab ON ab.inverter_id = ai.id LEFT JOIN ems.site_grid_connection sgc ON sgc.site_id = ai.site_id WHERE ai.site_id = $1 AND ai.active = true AND ai.controllable = true AND se.enabled = true AND se.endpoint_type = 'modbus_tcp' ORDER BY ai.id LIMIT 1 """, site_id, ) if row is None: return None mc = row["max_charge_a"] md = row["max_discharge_a"] max_charge_a = int(mc) if mc is not None else 0 max_discharge_a = int(md) if md is not None else 0 port = int(row["port"] or 502) uid = int(row["unit_id"] if row["unit_id"] is not None else 1) return InverterConfig( id=int(row["id"]), code=row["code"], host=row["host"], port=port, unit_id=uid, max_export_power_w=int(row["max_export_power_w"]) if row["max_export_power_w"] is not None else None, max_import_power_w=int(row["max_import_power_w"]) if row["max_import_power_w"] is not None else None, no_export=bool(row["no_export"] or False), max_battery_charge_w=int(row["max_battery_charge_w"]) if row["max_battery_charge_w"] is not None else None, max_battery_discharge_w=int(row["max_battery_discharge_w"]) if row["max_battery_discharge_w"] is not None else None, min_soc_percent=int(round(float(row["min_soc_percent"]))) if row["min_soc_percent"] is not None else None, reserve_soc_percent=int(row["reserve_soc_percent"]) if row["reserve_soc_percent"] is not None else None, max_soc_percent=int(row["max_soc_percent"]) if row["max_soc_percent"] is not None else None, usable_capacity_wh=int(row["usable_capacity_wh"]) if row["usable_capacity_wh"] is not None else None, max_charge_a=max_charge_a, max_discharge_a=max_discharge_a, deye_last_system_time_sync_minute=row["deye_last_system_time_sync_minute"], deye_last_system_time_sync_at=row["deye_last_system_time_sync_at"], deye_last_tou_inactive_write_prague_date=row[ "deye_last_tou_inactive_write_prague_date" ], deye_tou_inactive_signature=row["deye_tou_inactive_signature"], ) def _deye_system_time_register_rows() -> tuple[datetime, list[tuple[int, str, int]]]: """Hodnoty pro reg 62–64 (Europe/Prague); sekundy v reg 64 = 0 (stabilnější zápis).""" now = datetime.now(PRAGUE_TZ).replace(second=0, microsecond=0) reg62 = ((now.year - 2000) << 8) | now.month reg63 = (now.day << 8) | now.hour reg64 = (now.minute << 8) | 0 rows = [ (62, "", reg62), (63, "", reg63), (64, "", reg64), ] return now, rows def _deye_time_point_rows( slot_index: int, time_hhmm: int, power_w: int, soc_pct: int, grid_charge: bool, ) -> list[tuple[int, str, int]]: g = 1 if grid_charge else 0 return [ (148 + slot_index, "", time_hhmm), (154 + slot_index, "", power_w), (166 + slot_index, "", soc_pct), (172 + slot_index, "", g), ] def _slot_start_prague_sql(slot_offset: int) -> str: """Výraz TIMESTAMPTZ = začátek aktuálního (+offset) 15min slotu v Europe/Prague.""" off = int(slot_offset) return f""" ( WITH loc AS (SELECT now() AT TIME ZONE 'Europe/Prague' AS ts) SELECT ( (date_trunc('day', ts) + make_interval( hours => EXTRACT(HOUR FROM ts)::int, mins => (FLOOR(EXTRACT(MINUTE FROM ts) / 15) * 15)::int ) )::timestamp AT TIME ZONE 'Europe/Prague' ) + INTERVAL '{off * 15} minutes' FROM loc ) """ async def _fetch_plan_row_for_slot_offset( site_id: int, db: asyncpg.Connection, slot_offset: int ) -> asyncpg.Record | None: """Řádek plánu pro slot: 0 = probíhající 15min, 1 = následující (hranice v Europe/Prague).""" t = _slot_start_prague_sql(slot_offset) return await db.fetchrow( f""" SELECT pi.* FROM ems.planning_interval pi JOIN ems.planning_run pr ON pr.id = pi.run_id WHERE pr.site_id = $1 AND pr.status = 'active' AND pi.interval_start = {t} LIMIT 1 """, site_id, ) async def _fetch_max_charge_power_w(site_id: int, db: asyncpg.Connection) -> int: v = await db.fetchval( """ SELECT LEAST( COALESCE(ai.max_battery_charge_w, ai.max_charge_power_w), COALESCE( ab.bms_max_charge_w, CASE WHEN ab.max_charge_c_rate IS NOT NULL THEN (ab.max_charge_c_rate * ab.usable_capacity_wh)::bigint END, COALESCE(ai.max_battery_charge_w, ai.max_charge_power_w) ) ) AS effective_charge_w FROM ems.asset_battery ab JOIN ems.asset_inverter ai ON ai.id = ab.inverter_id AND ai.site_id = ab.site_id WHERE ab.site_id = $1 AND ai.controllable = true AND ai.active = true ORDER BY ab.id LIMIT 1 """, site_id, ) if v is None: return 0 return int(v) def _build_setpoints(mode: OperatingModeInfo, pi: asyncpg.Record | None) -> ControlSetpoints | None: code = mode.mode_code if code == "MANUAL": return None if code == "AUTO": if pi is None: return None grid_sp = int(pi["grid_setpoint_w"] or 0) ev1_w = int(pi["ev1_setpoint_w"] or 0) if "ev1_setpoint_w" in pi else 0 ev2_w = int(pi["ev2_setpoint_w"] or 0) if "ev2_setpoint_w" in pi else 0 hp_en = bool(pi["heat_pump_enabled"]) tgt = pi["battery_soc_target_pct"] target_soc = int(round(float(tgt))) if tgt is not None else None return ControlSetpoints( battery_w=int(pi["battery_setpoint_w"] or 0), grid_export_limit=abs(min(grid_sp, 0)), ev1_current_a=watts_to_amps(ev1_w, phases=3), ev2_current_a=watts_to_amps(ev2_w, phases=1), heat_pump_enable=hp_en, grid_setpoint_w=grid_sp, ev1_power_w=ev1_w, ev2_power_w=ev2_w, target_soc_pct=target_soc, ) if code == "SELF_SUSTAIN": return ControlSetpoints( battery_w=None, grid_export_limit=0, ev1_current_a=0, ev2_current_a=0, heat_pump_enable=False, grid_setpoint_w=0, ev1_power_w=0, ev2_power_w=0, target_soc_pct=None, ) if code == "CHARGE_CHEAP": # max_charge doplníme v export_setpoints z DB return ControlSetpoints( battery_w=0, grid_export_limit=0, ev1_current_a=0, ev2_current_a=0, heat_pump_enable=False, grid_setpoint_w=0, ev1_power_w=0, ev2_power_w=0, target_soc_pct=None, ) if code == "PRESERVE": return ControlSetpoints( battery_w=0, grid_export_limit=0, ev1_current_a=0, ev2_current_a=0, heat_pump_enable=False, grid_setpoint_w=0, ev1_power_w=0, ev2_power_w=0, target_soc_pct=None, lock_battery=True, ) logger.warning("Unknown mode_code %s for site export, skipping", code) return None def _apply_price_failsafe_guard( site_id: int, mode: OperatingModeInfo, pi: asyncpg.Record | None, sp: ControlSetpoints, ) -> ControlSetpoints: if mode.mode_code != "AUTO" or pi is None: return sp if "is_predicted_price" not in pi or not bool(pi["is_predicted_price"]): return sp logger.warning( "control export site=%s: AUTO slot uses predicted price -> forcing PASSIVE no-export guard", site_id, ) return ControlSetpoints( battery_w=0, grid_export_limit=0, ev1_current_a=sp.ev1_current_a, ev2_current_a=sp.ev2_current_a, heat_pump_enable=sp.heat_pump_enable, grid_setpoint_w=max(0, int(sp.grid_setpoint_w or 0)), ev1_power_w=sp.ev1_power_w, ev2_power_w=sp.ev2_power_w, target_soc_pct=sp.target_soc_pct, ) def _deye_reg143_export_w(no_export: bool, max_export_power_w: int | None) -> int: """Reg 143 – max export W z DB (např. SUN-20K / home-01 = 13 500 W).""" if no_export: return 0 return max(0, int(max_export_power_w or 0)) def _clamp_deye_tou_soc_pct(pct: int) -> int: return max(5, min(95, pct)) def _deye_tou_min_soc_pct(inv: InverterConfig) -> int: if inv.min_soc_percent is not None: return _clamp_deye_tou_soc_pct(int(inv.min_soc_percent)) return 10 def _deye_tou_reserve_soc_pct(inv: InverterConfig) -> int: if inv.reserve_soc_percent is not None: return _clamp_deye_tou_soc_pct(int(inv.reserve_soc_percent)) return 20 def get_deye_mode(setpoints: ControlSetpoints) -> str: """ Fyzický režim Deye: SELL | CHARGE | PASSIVE. Solver: záporný grid_setpoint_w = export; kladný výrazný + nabíjení = CHARGE ze sítě. battery_w=None (SELF_SUSTAIN) → bat_w považuj za 0 → typicky PASSIVE při grid_setpoint_w=0. """ grid_w = int(setpoints.grid_setpoint_w or 0) if setpoints.battery_w is None: bat_w = 0 else: bat_w = int(setpoints.battery_w) if grid_w < -200: return "SELL" if bat_w > 500 and grid_w > 200: return "CHARGE" return "PASSIVE" def _deye_tou_params( setpoints: ControlSetpoints, inv: InverterConfig, ) -> tuple[int, int, bool]: """ Parametry jednoho Deye time pointu: výkon W, SOC min %, grid_charge. Musí odpovídat logice get_deye_mode / lock_battery v write_inverter_setpoints. """ max_batt_w_discharge = int(inv.max_discharge_a * BATT_VOLTAGE_V) tp_discharge_w = 0 if setpoints.lock_battery else max_batt_w_discharge tou_min = _deye_tou_min_soc_pct(inv) tou_reserve = _deye_tou_reserve_soc_pct(inv) if setpoints.lock_battery: return tp_discharge_w, tou_min, False deye_mode = get_deye_mode(setpoints) if deye_mode == "CHARGE": raw_bat = setpoints.battery_w battery_w = int(raw_bat) if raw_bat is not None else 0 cap = int(inv.max_soc_percent) if inv.max_soc_percent is not None else 95 target_soc = max(10, min(95, cap)) tp_charge_w = battery_watts_to_amps(battery_w, inv.max_charge_a) * int(BATT_VOLTAGE_V) return tp_charge_w, target_soc, True if deye_mode == "SELL": return tp_discharge_w, tou_reserve, False return tp_discharge_w, tou_min, False async def write_inverter_setpoints( site_id: int, setpoints_now: ControlSetpoints, setpoints_next: ControlSetpoints | None, db: asyncpg.Connection, planning_run_id: int | None = None, ) -> str: inv = await _load_inverter_config(site_id, db) if inv is None: return "FAIL inverter: no controllable Modbus endpoint" raw_bat = setpoints_now.battery_w grid_w = int(setpoints_now.grid_setpoint_w or 0) no_export = inv.no_export export_lim = _deye_reg143_export_w(no_export, inv.max_export_power_w) max_batt_w_discharge = int(inv.max_discharge_a * BATT_VOLTAGE_V) tp_discharge_w = 0 if setpoints_now.lock_battery else max_batt_w_discharge tou_min_pct = _deye_tou_min_soc_pct(inv) tou_reserve_pct = _deye_tou_reserve_soc_pct(inv) try: soc_telemetry = await _get_current_soc(site_id, db) deye_mode = get_deye_mode(setpoints_now) if setpoints_now.lock_battery: charge_a = 0 discharge_a = 0 elif deye_mode == "CHARGE": battery_w = int(raw_bat) if raw_bat is not None else 0 charge_a = battery_watts_to_amps(battery_w, inv.max_charge_a) discharge_a = 0 else: charge_a = int(inv.max_charge_a) discharge_a = int(inv.max_discharge_a) selling_mode = 0 if deye_mode == "SELL" else 1 export_limit = export_lim reg178_val = REG178_SELL if deye_mode == "SELL" else REG178_PASSIVE logger.info( f"[control] site={site_id} fyzický režim Deye: {deye_mode} | " f"battery_w={raw_bat!r} grid_w={grid_w} | " f"charge_a={charge_a} discharge_a={discharge_a} | " f"reg142={'0=SELL' if deye_mode == 'SELL' else '1=ZERO_EXP'} " f"reg178={reg178_val}" ) now_cet, time_rows = _deye_system_time_register_rows() skip_time = False try: mb_clock = await get_modbus_client(inv.host, inv.port) tvals = await mb_clock.read_holding_registers( 62, 3, int(inv.unit_id if inv.unit_id is not None else 1) ) if len(tvals) == 3: skip_time = _deye_should_skip_time_sync_after_read( inv, int(tvals[0]), int(tvals[1]), int(tvals[2]) ) else: logger.warning( "Deye clock read: expected 3 registers, got %s; will sync 62–64", len(tvals), ) except Exception as e: logger.warning("Deye clock read failed (will sync 62–64): %s", e) if skip_time: logger.info( "Deye clock 62–64 skipped (drift ≤ %ss, last write < %sh ago): %s CET", DEYE_CLOCK_DRIFT_OK_SEC, DEYE_CLOCK_RESYNC_INTERVAL_HOURS, now_cet.strftime("%Y-%m-%d %H:%M:%S"), ) else: logger.info("Deye time will sync: %s CET", now_cet.strftime("%Y-%m-%d %H:%M:%S")) registers: list[tuple[int, str, int]] = [] if skip_time else list(time_rows) sp_tp2 = setpoints_next if setpoints_next is not None else setpoints_now hh_cur = current_slot_hhmm() hh_nxt = next_slot_hhmm() p1, s1, g1 = _deye_tou_params(setpoints_now, inv) p2, s2, g2 = _deye_tou_params(sp_tp2, inv) registers.extend(_deye_time_point_rows(0, hh_cur, p1, s1, g1)) registers.extend(_deye_time_point_rows(1, hh_nxt, p2, s2, g2)) prague_date = datetime.now(PRAGUE_TZ).date() inactive_sig = ( f"{DEYE_TOU_INACTIVE_HHMM}|{tou_min_pct}|{tou_reserve_pct}|{tp_discharge_w}" ) need_inactive_tou = ( inv.deye_last_tou_inactive_write_prague_date != prague_date or inv.deye_tou_inactive_signature != inactive_sig ) if need_inactive_tou: for idx in range(2, 6): registers.extend( _deye_time_point_rows( idx, DEYE_TOU_INACTIVE_HHMM, tp_discharge_w, tou_min_pct, False ) ) else: logger.debug( "Deye TOU rows 3–6 skipped (already written today, signature unchanged)" ) registers.extend( [ (108, "", charge_a), (109, "", discharge_a), (141, "energy_mode (0)", 0), (142, "limit_control (0=selling, 1=zero_export)", selling_mode), (178, "grid_peak_shaving_switch", reg178_val), (143, "", export_limit), ] ) logger.info( "[control] %s: deye_mode=%s charge=%sA discharge=%sA limit_control=%s export=%sW " "time_point1=%s time_point2=%s soc_telemetry=%s%% (batt=%r grid=%sW)", inv.code, deye_mode, charge_a, discharge_a, selling_mode, export_limit, hh_cur, hh_nxt, soc_telemetry, raw_bat, grid_w, ) last_verified = await _fetch_last_verified_inverter_registers(site_id, inv.id, db) registers, skipped_unchanged = _drop_registers_matching_last_verified( registers, last_verified ) if skipped_unchanged: logger.info( "[control] %s: skip %s registers (value equals last verified): %s", inv.code, len(skipped_unchanged), skipped_unchanged[:24], ) if not registers: logger.info( "[control] %s: all Deye holding regs match last verified, no Modbus write", inv.code, ) if need_inactive_tou: await db.execute( """ UPDATE ems.asset_inverter SET deye_last_tou_inactive_write_prague_date = $1, deye_tou_inactive_signature = $2 WHERE id = $3 """, prague_date, inactive_sig, inv.id, ) return ( f"OK inverter: batt_w={raw_bat!r} (no changes vs last verified Modbus snapshot)" ) will_write_inactive = any( int(r) in _DEYE_INACTIVE_TOU_REGISTERS for r, _, _ in registers ) cmd_ids = await create_modbus_commands( site_id, planning_run_id, "inverter", inv.id, inv.code, inv.host, inv.port, inv.unit_id, registers, db, deye_physical_mode=deye_mode, ) if not await execute_modbus_commands(cmd_ids, db): return f"FAIL inverter: {inv.code}: Modbus write failed (see modbus_command)" logger.info("[control] Inverter %s journal write OK", inv.code) if need_inactive_tou or will_write_inactive: await db.execute( """ UPDATE ems.asset_inverter SET deye_last_tou_inactive_write_prague_date = $1, deye_tou_inactive_signature = $2 WHERE id = $3 """, prague_date, inactive_sig, inv.id, ) except Exception as e: return f"FAIL inverter: {inv.code}: {e}" return ( f"OK inverter: batt_w={raw_bat!r} " f"(time points + FC 0x10: 108/109/141/142/178/143)" ) async def read_deye_registers_live(site_id: int, db: asyncpg.Connection) -> dict[str, Any]: """ Živé čtení holding registrů Deye 108, 109, 141, 142, 143, 178, 191 (stejné TCP spojení jako telemetrie/export). Vše pod jedním mutexem + sdružené FC3 bloky — mezi jednotlivými read_register dřív telemetrie střídavě brala lock a RS485 brány házely cizí transaction_id / I/O timeouty. """ inv = await _load_inverter_config(site_id, db) if inv is None: raise ValueError("no controllable Modbus inverter for site") uid = int(inv.unit_id) client = await get_modbus_client(inv.host, inv.port) read_at = datetime.now(timezone.utc) try: async with client.batch(uid) as mb: b108 = await mb.read_holding_registers(108, 2) b141 = await mb.read_holding_registers(141, 3) r178 = await mb.read_holding_registers(178, 1) r191 = await mb.read_holding_registers(191, 1) r108, r109 = b108[0], b108[1] r141, r142, r143 = b141[0], b141[1], b141[2] r178 = r178[0] r191 = r191[0] except Exception: logger.exception("read_deye_registers_live site=%s failed", site_id) raise return { "reg108_charge_a": int(r108), "reg109_discharge_a": int(r109), "reg141_energy_mode": int(r141), "reg142_limit_control": int(r142), "reg143_export_limit_w": int(r143), "reg178_peak_shaving_switch": int(r178), "reg191_peak_shaving_w": int(r191), "read_at": read_at.isoformat(), } def _current_limit_for_charger(charger_code: str, sp: ControlSetpoints) -> int: c = (charger_code or "").strip().lower() if c == "ev-charger-1": a = sp.ev1_current_a elif c == "ev-charger-2": a = sp.ev2_current_a elif c.endswith("-1") or c == "ev1": a = sp.ev1_current_a elif c.endswith("-2") or c == "ev2": a = sp.ev2_current_a else: a = 0 if a < 6: a = 0 return a async def write_ev_setpoints(site_id: int, setpoints: ControlSetpoints, db: asyncpg.Connection) -> str: rows = await db.fetch( """ SELECT ec.code, se.host, se.port, se.unit_id FROM ems.asset_ev_charger ec JOIN ems.site_endpoint se ON se.id = ec.endpoint_id WHERE ec.site_id = $1 AND ec.schedulable = true AND se.enabled = true AND se.endpoint_type = 'modbus_tcp' ORDER BY ec.code """, site_id, ) if not rows: return "OK EV: no schedulable chargers" for row in rows: code = row["code"] current_a = _current_limit_for_charger(code, setpoints) logger.info( "EV setpoint [%s]: %sA (TODO: Modbus registers)", code, current_a, ) return f"OK EV: logged {len(rows)} charger(s) (Modbus TODO)" async def write_heat_pump_setpoint(site_id: int, setpoints: ControlSetpoints, db: asyncpg.Connection) -> str: rows = await db.fetch( """ SELECT hp.code, se.host, se.port, se.unit_id FROM ems.asset_heat_pump hp JOIN ems.site_endpoint se ON se.id = hp.endpoint_id WHERE hp.site_id = $1 AND hp.schedulable = true AND se.enabled = true AND se.endpoint_type = 'modbus_tcp' """, site_id, ) if not rows: return "OK heat pump: no schedulable unit" for row in rows: logger.info( "HP setpoint [%s]: enable=%s (TODO: Modbus registers)", row["code"], setpoints.heat_pump_enable, ) return "OK heat pump: logged (Modbus TODO)" async def send_loxone_setpoints( site_id: int, setpoints: ControlSetpoints, mode: OperatingModeInfo, db: asyncpg.Connection, ) -> str: endpoint = await db.fetchrow( """ SELECT host, port, protocol FROM ems.site_endpoint WHERE site_id = $1 AND endpoint_type = 'loxone_http' AND enabled = true ORDER BY id LIMIT 1 """, site_id, ) if not endpoint: return "OK Loxone: no endpoint, skipped" proto = (endpoint["protocol"] or "http").lower() if proto not in ("http", "https"): proto = "http" host = endpoint["host"] port = int(endpoint["port"] or (443 if proto == "https" else 80)) base = f"{proto}://{host}:{port}/dev/sps/io" settings = get_settings() user = settings.loxone_user or os.getenv("LOXONE_USER") or "" password = settings.loxone_password or os.getenv("LOXONE_PASSWORD") or "" auth = (user, password) if user else None batt_display = 0 if setpoints.battery_w is None else int(setpoints.battery_w) paths: list[tuple[str, int]] = [ (f"{base}/EMS_Mode/{mode.loxone_mode_value}", mode.loxone_mode_value), (f"{base}/EMS_Battery_Setpoint_W/{batt_display}", batt_display), (f"{base}/EMS_Grid_Setpoint_W/{setpoints.grid_setpoint_w}", setpoints.grid_setpoint_w), (f"{base}/EMS_EV1_Power_W/{setpoints.ev1_power_w}", setpoints.ev1_power_w), (f"{base}/EMS_EV2_Power_W/{setpoints.ev2_power_w}", setpoints.ev2_power_w), (f"{base}/EMS_HeatPump_Enable/{1 if setpoints.heat_pump_enable else 0}", 1 if setpoints.heat_pump_enable else 0), ] errs: list[str] = [] try: async with httpx.AsyncClient(timeout=5.0) as client: for url, _ in paths: try: r = await client.get(url, auth=auth) r.raise_for_status() except Exception as e: errs.append(f"{url!s}: {e}") except Exception as e: return f"FAIL Loxone: client {e}" if errs: return "FAIL Loxone: " + "; ".join(errs[:3]) return "OK Loxone: all virtual inputs updated" async def export_setpoints(site_id: int, db: asyncpg.Connection) -> None: mode = await _fetch_operating_mode(site_id, db) if mode is None: logger.warning("control export site=%s: no operating mode row", site_id) return if mode.mode_code == "MANUAL": logger.info("control export site=%s: MANUAL, skip writes", site_id) return pi_now = await _fetch_plan_row_for_slot_offset(site_id, db, 0) pi_next = await _fetch_plan_row_for_slot_offset(site_id, db, 1) sp_now = _build_setpoints(mode, pi_now) sp_next = _build_setpoints(mode, pi_next) if mode.mode_code == "AUTO" and sp_now is None: if pi_now is None: logger.warning( "control export site=%s: AUTO but no planning_interval for current slot, skip", site_id, ) return if sp_now is None: logger.warning( "control export site=%s: no setpoints for mode %s, skip", site_id, mode.mode_code, ) return if mode.mode_code == "CHARGE_CHEAP": max_ch = await _fetch_max_charge_power_w(site_id, db) # Kladný grid_setpoint_w > 200 → fyzický CHARGE (nabíjení ze sítě), viz get_deye_mode grid_for_charge = max(300, max_ch) sp_now = ControlSetpoints( battery_w=max_ch, grid_export_limit=0, ev1_current_a=0, ev2_current_a=0, heat_pump_enable=False, grid_setpoint_w=grid_for_charge, ev1_power_w=0, ev2_power_w=0, target_soc_pct=None, ) sp_next = sp_now else: sp_now = _apply_price_failsafe_guard(site_id, mode, pi_now, sp_now) if sp_next is not None: sp_next = _apply_price_failsafe_guard(site_id, mode, pi_next, sp_next) planning_run_id = await db.fetchval( """ SELECT id FROM ems.planning_run WHERE site_id = $1 AND status = 'active' ORDER BY created_at DESC LIMIT 1 """, site_id, ) if planning_run_id is not None: planning_run_id = int(planning_run_id) try: inv_res = await write_inverter_setpoints( site_id, sp_now, sp_next, db, planning_run_id=planning_run_id ) except Exception as e: logger.error("inverter write failed: %s", e) inv_res = f"FAIL inverter: {e}" try: ev_res = await write_ev_setpoints(site_id, sp_now, db) except Exception as e: logger.error("ev write failed: %s", e) ev_res = f"FAIL ev: {e}" try: hp_res = await write_heat_pump_setpoint(site_id, sp_now, db) except Exception as e: logger.error("hp write failed: %s", e) hp_res = f"FAIL heat pump: {e}" try: lox_res = await send_loxone_setpoints(site_id, sp_now, mode, db) except Exception as e: logger.error("loxone write failed: %s", e) lox_res = f"FAIL Loxone: {e}" results = list( zip( ("inverter", "ev", "heat_pump", "loxone"), (inv_res, ev_res, hp_res, lox_res), ) ) for name, res in results: if isinstance(res, Exception): logger.error("control export site=%s %s: FAIL %s", site_id, name, res) elif isinstance(res, str) and res.startswith("FAIL"): logger.error("control export site=%s %s: %s", site_id, name, res) else: logger.info("control export site=%s %s: %s", site_id, name, res)