|
| 1 | +"""Exercise the DEYE command translator against an isolated HA state copy.""" |
| 2 | + |
| 3 | +from copy import deepcopy |
| 4 | +from datetime import datetime, timezone |
| 5 | +from types import SimpleNamespace |
| 6 | + |
| 7 | +from deye_local import DeyeLocal |
| 8 | + |
| 9 | + |
| 10 | +class DeyeShadowState: |
| 11 | + """A narrow in-memory interface with no route to a real control service.""" |
| 12 | + |
| 13 | + def __init__(self, base, settings): |
| 14 | + """Copy the required observations once; never retain the real base object.""" |
| 15 | + self.prefix = base.prefix |
| 16 | + self.set_read_only = False |
| 17 | + self.set_read_only_axle = False |
| 18 | + self.manual_soc_max_keep = deepcopy(base.manual_soc_max_keep) |
| 19 | + self.timezone = base.get_arg("timezone", "Europe/London") |
| 20 | + self.states = {} |
| 21 | + self.writes = [] |
| 22 | + self.report = {} |
| 23 | + self.errors = [] |
| 24 | + self.allowed = {row[field] for row in settings["programs"] for field in ("soc", "power", "charging")} |
| 25 | + entities = {entity for row in settings["programs"] for entity in row.values()} |
| 26 | + entities.update(settings[key] for key in ("time_of_use", "work_mode", "energy_pattern", "telemetry_time")) |
| 27 | + for entity in entities: |
| 28 | + self.states[entity] = deepcopy(base.get_state_wrapper(entity, raw=True)) |
| 29 | + |
| 30 | + def get_arg(self, name, default=None): |
| 31 | + """Expose only the timezone needed to parse the existing clock reading.""" |
| 32 | + return self.timezone if name == "timezone" else default |
| 33 | + |
| 34 | + def get_state_wrapper(self, entity, raw=False): |
| 35 | + """Read a copied observation, not Home Assistant or a device.""" |
| 36 | + return deepcopy(self.states.get(entity)) |
| 37 | + |
| 38 | + def call_service_wrapper(self, service, **data): |
| 39 | + """Record the proposed write and simulate acceptance only in this copy.""" |
| 40 | + entity = data["entity_id"] |
| 41 | + if entity not in self.allowed or service not in ("number/set_value", "select/select_option"): |
| 42 | + raise ValueError("Unexpected DEYE shadow control") |
| 43 | + value = data["value"] if service == "number/set_value" else data["option"] |
| 44 | + self.writes.append({"service": service, "entity_id": entity, "value": value}) |
| 45 | + self.states[entity]["state"] = value |
| 46 | + return True |
| 47 | + |
| 48 | + def record_status(self, message, had_errors=False): |
| 49 | + """Retain isolated failures without changing the native planner's status.""" |
| 50 | + self.errors.append(message) |
| 51 | + |
| 52 | + def dashboard_item(self, entity, state, attributes): |
| 53 | + """Capture adapter output without publishing intermediate simulation states.""" |
| 54 | + self.report = {"state": state, **attributes} |
| 55 | + |
| 56 | + |
| 57 | +def preview_deye_local(inverter, target_soc, freeze=False): |
| 58 | + """Publish a non-executing trace alongside the unchanged native controller. |
| 59 | +
|
| 60 | + Success means the translator completed against simulated acknowledgements, |
| 61 | + not that the real inverter accepted or executed anything. No capabilities, |
| 62 | + manual targets, windows, rate intents or actual observations are changed. |
| 63 | + """ |
| 64 | + base = inverter.base |
| 65 | + settings = base.args.get("deye_shadow") |
| 66 | + if settings is None: |
| 67 | + return None |
| 68 | + report = { |
| 69 | + "generated_at": datetime.now(timezone.utc).isoformat(), |
| 70 | + "execution_enabled": False, |
| 71 | + "actuator_signals_modified": False, |
| 72 | + "ready_to_activate": False, |
| 73 | + "hardware_confirmed": False, |
| 74 | + "target_soc": target_soc, |
| 75 | + "freeze": bool(freeze), |
| 76 | + "native_inverter_type": inverter.inverter_type, |
| 77 | + "simulated_acknowledgements": True, |
| 78 | + "proposed_writes": [], |
| 79 | + } |
| 80 | + state = "blocked" |
| 81 | + try: |
| 82 | + if not isinstance(settings, dict) or settings.get("control", False) is not False: |
| 83 | + raise ValueError("deye_shadow cannot enable physical control") |
| 84 | + settings = DeyeLocal.validate_settings({**settings, "control": True}) |
| 85 | + simulated = DeyeShadowState(base, settings) |
| 86 | + fake_inverter = SimpleNamespace( |
| 87 | + base=simulated, |
| 88 | + reserve_percent=inverter.reserve_percent, |
| 89 | + battery_rate_max_discharge=inverter.battery_rate_max_discharge, |
| 90 | + inv_write_and_poll_sleep=0, |
| 91 | + count_register_writes=0, |
| 92 | + ) |
| 93 | + adapter = DeyeLocal(fake_inverter) |
| 94 | + adapter.settings = settings |
| 95 | + success = adapter.apply(target_soc, freeze) |
| 96 | + state = "simulated" if success else "blocked" |
| 97 | + report["reason"] = simulated.report.get("reason", "Physical commissioning and executable integration remain required") |
| 98 | + report["mode"] = simulated.report.get("mode", "none") |
| 99 | + report["proposed_writes"] = deepcopy(simulated.writes) |
| 100 | + report["proposed_write_count"] = len(simulated.writes) |
| 101 | + report["hold_verified"] = settings.get("hold_verified", False) |
| 102 | + if not success and "boundaries must be strictly increasing" in report["reason"]: |
| 103 | + # Keep the REAL result blocked. Also exercise the same translator |
| 104 | + # under an explicitly labelled table-repair hypothesis, so a legacy |
| 105 | + # two-boundary SK mapping does not prevent useful offline traces. |
| 106 | + times = [simulated.states[row["time"]]["state"] for row in settings["programs"]] |
| 107 | + minutes = [int(stamp[:2]) * 60 + int(stamp[3:5]) for stamp in times] |
| 108 | + if len(set(minutes)) == 6: |
| 109 | + ordered = [stamp for _, stamp in sorted(zip(minutes, times))] |
| 110 | + for row, stamp in zip(settings["programs"], ordered): |
| 111 | + simulated.states[row["time"]]["state"] = stamp |
| 112 | + simulated.writes.clear() |
| 113 | + repaired_success = adapter.apply(target_soc, freeze) |
| 114 | + report["table_repair_hypothesis"] = { |
| 115 | + "physical_table_changed": False, |
| 116 | + "required_before_activation": True, |
| 117 | + "ordered_start_times": ordered, |
| 118 | + "translation_succeeded": repaired_success, |
| 119 | + "proposed_writes": simulated.writes, |
| 120 | + } |
| 121 | + except Exception as exc: |
| 122 | + # A preview must never interrupt SK's actual charge/stop action. |
| 123 | + report["reason"] = str(exc) if isinstance(exc, ValueError) else "DEYE shadow input or simulation error" |
| 124 | + base.dashboard_item(base.prefix + ".deye_shadow", state=state, attributes=report) |
| 125 | + return report |
0 commit comments