Files
ems/backend/services/control/exporter_monolith.py
2026-05-02 19:51:41 +02:00

560 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Export plánovaných setpointů na Modbus (Deye, EV, TČ) a HTTP do Loxone."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Any
import asyncpg
from services.control.deye_helpers import (
BATT_VOLTAGE_V,
DEYE_CLOCK_DRIFT_OK_SEC,
DEYE_CLOCK_REGS,
DEYE_CLOCK_RESYNC_INTERVAL_HOURS,
DEYE_CLOCK_VERIFY_MAX_DELTA_SEC, # noqa: F401 - re-export for compatibility
DEYE_CRITICAL_REGS_SELF_SUSTAIN, # noqa: F401 - re-export for compatibility
DEYE_REGISTER_NAMES, # noqa: F401 - re-export for compatibility
DEYE_TOU_INACTIVE_HHMM,
DEYE_TOU_POWER_REGS,
PRAGUE_TZ,
REG143_SELL_CAP_MIN_W,
REG178_MI_EXPORT_DISABLE,
REG178_MI_EXPORT_ENABLE,
REG178_MI_EXPORT_MASK,
REG178_PASSIVE,
REG178_SELL,
REG178_VERIFY_MASK,
REG178_VERIFY_MASK_COMBINED,
_DEYE_INACTIVE_TOU_REGISTERS,
_deye_clock_registers_verify_match,
_deye_reg178_verify_match,
_deye_reg178_verify_with_double_read,
_deye_registers_to_prague_datetime, # noqa: F401 - re-export for compatibility
_deye_should_skip_time_sync_after_read,
_deye_tou_power_verify_match,
_prague_minute_start_utc,
battery_watts_to_amps,
compute_pv_a_reg340_max_solar_w,
current_slot_hhmm,
deye_reg_triggers_self_sustain_after_verify_exhaust, # noqa: F401 - re-export
next_slot_hhmm,
watts_to_amps,
)
from services.control.models import ControlSetpoints, InverterConfig, OperatingModeInfo
from services.control.modbus_journal import (
_drop_registers_matching_last_verified,
_fetch_last_verified_inverter_registers,
create_modbus_commands,
execute_modbus_commands,
)
from services.control.outputs import (
_current_limit_for_charger,
send_loxone_setpoints,
write_ev_setpoints,
write_heat_pump_setpoint,
)
from services.control.repository import (
_fetch_max_charge_power_w,
_fetch_operating_mode,
_fetch_plan_row_for_slot_offset,
_get_current_soc,
_load_inverter_config,
)
from services.control.setpoints import (
_DictRecord,
_apply_price_failsafe_guard,
_build_setpoints,
_clamp_deye_tou_soc_pct,
_deye_passive_tou_battery_soc_pct,
_deye_reg143_export_w,
_deye_system_time_register_rows,
_deye_time_point_rows,
_deye_tou_min_soc_pct,
_deye_tou_params,
_deye_tou_reserve_soc_pct,
_deye_zero_export_amps_for_passive,
get_deye_mode,
)
from services.control.verify import (
_deye_expected_clock_triplet_for_verify,
_modbus_cmd_register,
_switch_to_self_sustain,
_verify_deye_clock_written_bundle,
verify_modbus_commands,
)
from services.modbus_client import get_modbus_client
from services.signal_service import enqueue_site_signals
logger = logging.getLogger(__name__)
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"
unit_id = int(inv.unit_id if inv.unit_id is not None else 1)
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)
bat_w = int(raw_bat) if raw_bat is not None else 0
if setpoints_now.lock_battery:
charge_a = 0
discharge_a = 0
elif deye_mode == "CHARGE":
charge_a = battery_watts_to_amps(bat_w, inv.max_charge_a)
discharge_a = 0
elif deye_mode == "SELL":
# Záměrný výdej baterie do sítě: plný vybíjecí proud; export strop dle plánu níže.
charge_a = 0
discharge_a = int(inv.max_discharge_a)
elif setpoints_now.self_sustain_local_use:
# SELF_SUSTAIN: plný nabíjecí i vybíjecí proud invertoru — přebytek FVE jde do baterie,
# reg. 142 = zero export to load/CT (viz selling_mode níže), ne reg. 108 = 0.
charge_a = int(inv.max_charge_a)
discharge_a = int(inv.max_discharge_a)
else:
# PASSIVE (ZERO): výchozí plné 108/109; u přetoku FVE do sítě nebo importu bez baterie viz helper.
charge_a, discharge_a = _deye_zero_export_amps_for_passive(
grid_w,
bat_w,
int(inv.max_charge_a),
int(inv.max_discharge_a),
)
zero_exp_mode = int(inv.deye_zero_export_mode or 1)
selling_mode = 0 if deye_mode == "SELL" else zero_exp_mode
solar_sell = 0 if (setpoints_now.export_ban and deye_mode != "SELL") else 1
export_limit = export_lim
if deye_mode == "SELL" and grid_w < 0:
export_limit = min(export_lim, max(REG143_SELL_CAP_MIN_W, abs(grid_w)))
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={selling_mode} reg145={solar_sell} reg143={export_limit}W 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 6264",
len(tvals),
)
except Exception as e:
logger.warning("Deye clock read failed (will sync 6264): %s", e)
if skip_time:
logger.info(
"Deye clock 6264 skipped (drift ≤ %ss, last sync < %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 36 skipped (already written today, signature unchanged)"
)
registers.extend(
[
(108, "", charge_a),
(109, "", discharge_a),
(141, "energy_mode (0)", 0),
(142, "limit_control", selling_mode),
(143, "", export_limit),
(145, "solar_sell", solar_sell),
]
)
if (
bool(inv.deye_reg340_pv_a_control_enabled)
and int(inv.pv_a_cap_w) > 0
and setpoints_now.pv_a_allowed_w is not None
):
registers.append((340, "max_solar_power_w", int(setpoints_now.pv_a_allowed_w)))
# Reg 178: bitové pole. Nastavujeme bits45 (peak shaving) vždy; bits01 (MI export cutoff) jen pokud feature.
# Ostatní bity musí zůstat zachované → read-modify-write.
try:
mb178 = await get_modbus_client(inv.host, inv.port)
r178 = await mb178.read_holding_registers(178, 1, unit_id)
if not r178 or len(r178) < 1:
raise OSError(f"reg178 read returned {len(r178) if r178 is not None else None} values")
current_178 = int(r178[0])
peak_bits = int(reg178_val) & int(REG178_VERIFY_MASK)
if inv.deye_gen_microinverter_cutoff_enabled:
want_cutoff = bool(setpoints_now.deye_gen_cutoff_enabled) and deye_mode != "SELL"
mi_bits = (
REG178_MI_EXPORT_ENABLE if want_cutoff else REG178_MI_EXPORT_DISABLE
)
else:
mi_bits = int(current_178) & int(REG178_MI_EXPORT_MASK)
new_178 = (
(int(current_178) & ~int(REG178_VERIFY_MASK_COMBINED))
| int(peak_bits)
| int(mi_bits)
)
registers.append((178, "control_board_special_1", int(new_178)))
logger.info(
"[control] %s: reg178 (control_board_special_1) old=%s new=%s peak_bits=0x%04X mi_bits=%s",
inv.code,
current_178,
new_178,
int(peak_bits),
int(mi_bits),
)
except Exception as e:
logger.warning("[control] %s: reg178 RMW failed (skip reg178 write): %s", inv.code, e)
logger.info(
"[control] %s: deye_mode=%s charge=%sA discharge=%sA "
"reg142=%s reg145=%s export=%sW "
"tp1=%s tp2=%s soc=%s%% (batt=%r grid=%sW)",
inv.code,
deye_mode,
charge_a,
discharge_a,
selling_mode,
solar_sell,
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)
will_write_time = any(int(r) in DEYE_CLOCK_REGS for r, _, _ in registers)
if will_write_time:
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.id,
)
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/145/340 dle plánu)"
)
async def read_deye_registers_live(site_id: int, db: asyncpg.Connection) -> dict[str, Any]:
"""
Živé čtení holding registrů Deye 108, 109, 141145, 178, 191 a volitelně 340
(jen pokud `deye_reg340_pv_a_control_enabled`, jinak `reg340_max_solar_power_w` = null).
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, 5)
r178 = await mb.read_holding_registers(178, 1)
r191 = await mb.read_holding_registers(191, 1)
if inv.deye_reg340_pv_a_control_enabled:
r340 = await mb.read_holding_registers(340, 1)
else:
r340 = None
r108, r109 = b108[0], b108[1]
r141, r142, r143 = b141[0], b141[1], b141[2]
r145 = b141[4]
r178 = r178[0]
r191 = r191[0]
r340v = (
int(r340[0])
if r340 is not None and len(r340) >= 1
else None
)
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),
"reg145_solar_sell": int(r145),
"reg178_peak_shaving_switch": int(r178),
"reg178_control_board_special_1": int(r178),
"reg178_mi_export_cutoff_bits": int(r178) & int(REG178_MI_EXPORT_MASK),
"reg178_mi_export_cutoff_is_on": (int(r178) & int(REG178_MI_EXPORT_MASK)) == int(REG178_MI_EXPORT_ENABLE),
"reg191_peak_shaving_w": int(r191),
"reg340_max_solar_power_w": r340v,
"read_at": read_at.isoformat(),
}
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
try:
inv_for_pv = await _load_inverter_config(site_id, db)
cap_pv = int(inv_for_pv.pv_a_cap_w) if inv_for_pv is not None else 0
reg340_en = (
bool(inv_for_pv.deye_reg340_pv_a_control_enabled)
if inv_for_pv is not None
else False
)
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,
pv_a_cap_w=cap_pv,
reg340_pv_a_control_enabled=reg340_en,
)
sp_next = _build_setpoints(
mode,
pi_next,
pv_a_cap_w=cap_pv,
reg340_pv_a_control_enabled=reg340_en,
)
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)
# Oba setpointy kladné → get_deye_mode CHARGE; min. 1 W, aby režim nebyl PASSIVE při nulové DB.
pw = max(1, int(max_ch))
sp_now = ControlSetpoints(
battery_w=pw,
grid_export_limit=0,
ev1_current_a=0,
ev2_current_a=0,
heat_pump_enable=False,
grid_setpoint_w=pw,
ev1_power_w=0,
ev2_power_w=0,
target_soc_pct=None,
effective_sell_price_czk_kwh=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)
finally:
try:
await enqueue_site_signals(site_id, db)
except Exception as e:
logger.warning(
"control export site=%s: signal enqueue failed: %s", site_id, e
)