#!/usr/bin/env python3 """Bisect Infeasible na reálných slotech home-01 (fixture z MCP). Export fixture z MCP (server user-postgres-ems, nástroj query): python scripts/diagnose_home01_infeasible.py --print-export-sql --run-id 23784 # výstup SQL vlož do MCP query; JSON ulož např.: # scripts/home01_run23784_slots.json Spuštění bisectu: PYTHONPATH=backend python scripts/diagnose_home01_infeasible.py \\ --fixture scripts/home01_run23784_slots.json --soc-wh 51840 """ from __future__ import annotations import argparse import json import sys from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backend")) from services.planning_engine import ( # noqa: E402 PLANNER_BUILD_TAG, PlanningSlot, SOLVER_RELAX_STEPS, solve_dispatch, solve_dispatch_two_pass, ) DEFAULT_FIXTURE = Path(__file__).with_name("home01_run16674_slots.json") DEFAULT_SOC_WH = 37120.0 def export_slots_sql(run_id: int) -> str: """SQL pro MCP export slotů z fn_load_planning_slots_full pro daný run.""" return f""" select json_agg(row order by row.interval_start) from ( select s.interval_start, s.buy_price::float8 as buy, s.sell_price::float8 as sell, s.load_baseline_w as load, s.pv_a_forecast_w as pv_a, s.pv_b_forecast_w as pv_b, s.allow_charge, s.allow_discharge_export from ems.planning_run pr cross join lateral ems.fn_load_planning_slots_full( pr.site_id, coalesce(pr.replan_from, pr.horizon_start), pr.horizon_end, coalesce(pr.soc_at_replan_wh, 0) ) s where pr.id = {run_id} ) row; """.strip() def _ctx() -> tuple[SimpleNamespace, SimpleNamespace, SimpleNamespace, list]: battery = SimpleNamespace( usable_capacity_wh=64000.0, min_soc_wh=6400.0, arb_floor_wh=12800.0, reserve_soc_wh=12800.0, soc_max_wh=64000.0, charge_efficiency=0.95, discharge_efficiency=0.95, degradation_cost_czk_kwh=0.15, max_charge_power_w=18000, max_discharge_power_w=18000, charge_slot_buffer=1.3, discharge_slot_buffer=1.5, planner_terminal_soc_value_factor=0.9, planner_discharge_floor_percent=5.0, planner_extreme_buy_threshold_czk_kwh=-2.0, planner_daytime_charge_target_enabled=True, planner_charge_commitment_penalty_czk_kwh=0.2, planner_night_baseload_buffer_percent=20, planner_neg_sell_prep_soc_percent=10.0, planner_neg_sell_full_soc_tail_slots=4, planner_neg_sell_vent_min_sell_czk_kwh=-0.5, ) grid = SimpleNamespace( max_import_power_w=17000, max_export_power_w=13500, block_export_on_negative_sell=False, deye_gen_microinverter_cutoff_enabled=False, purchase_pricing_mode="spot", ) hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0) vehicles = [ SimpleNamespace(max_charge_power_w=11000, battery_capacity_kwh=75.0, default_target_soc_pct=80.0), SimpleNamespace(max_charge_power_w=7400, battery_capacity_kwh=52.0, default_target_soc_pct=90.0), ] return battery, hp, grid, vehicles def load_slots( rows: list[dict], *, permissive_masks: bool, use_row_masks: bool, ) -> list[PlanningSlot]: out: list[PlanningSlot] = [] for r in rows: ts = datetime.fromisoformat(r["interval_start"].replace("Z", "+00:00")) pv_surplus = max(0, int(r["pv_a"]) + int(r["pv_b"]) - int(r["load"])) if permissive_masks: allow_charge = True allow_discharge_export = True elif use_row_masks and "allow_charge" in r: allow_charge = bool(r.get("allow_charge")) allow_discharge_export = bool(r.get("allow_discharge_export", True)) else: allow_charge = float(r["buy"]) < 0 or ( float(r["sell"]) < 0 and pv_surplus > 500 ) allow_discharge_export = float(r["sell"]) >= 0 out.append( PlanningSlot( interval_start=ts, buy_price=float(r["buy"]), sell_price=float(r["sell"]), pv_a_forecast_w=int(r["pv_a"]), pv_b_forecast_w=int(r["pv_b"]), load_baseline_w=int(r["load"]), ev1_connected=False, ev2_connected=False, allow_charge=allow_charge, allow_discharge_export=allow_discharge_export, ) ) return out def try_solve( label: str, slots: list[PlanningSlot], soc_wh: float, **kwargs, ) -> tuple[str, dict | None]: battery, hp, grid, vehicles = _ctx() try: if kwargs.pop("two_pass", False): _results, _ms, snap = solve_dispatch_two_pass( slots, battery, hp, grid, [None, None], vehicles, soc_wh, 55.0, operating_mode="AUTO", **kwargs, ) else: _results, _ms, snap = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc_wh, 55.0, operating_mode="AUTO", **kwargs, ) inp = snap.get("inputs") or {} relax = inp.get("relax_chain") or [] push = len(inp.get("evening_push_ts") or []) suppressed = inp.get("evening_push_hard_suppressed") return ( f"OK {label} relax={relax[-1] if relax else 'strict'} " f"push_slots={push} hard_suppressed={suppressed}", inp, ) except Exception as e: return f"FAIL {label}: {e}", None def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) parser.add_argument("--soc-wh", type=float, default=DEFAULT_SOC_WH) parser.add_argument("--run-id", type=int, help="MCP run id pro --print-export-sql") parser.add_argument( "--print-export-sql", action="store_true", help="Vytiskne SQL pro export slotů z MCP", ) args = parser.parse_args() if args.print_export_sql: run_id = args.run_id or 23784 print(export_slots_sql(run_id)) print( f"\n-- Ulož json_agg výsledek do {args.fixture.name} " f"(nebo jiné cesty přes --fixture).", file=sys.stderr, ) return fixture = args.fixture if not fixture.exists(): print( f"Chybí {fixture}. Spusť:\n" f" python {Path(__file__).name} --print-export-sql --run-id 23784", file=sys.stderr, ) sys.exit(1) rows = json.loads(fixture.read_text()) neg_buy = [r for r in rows if r["buy"] < 0] neg_sell = [r for r in rows if r["sell"] < 0] print("tag", PLANNER_BUILD_TAG) print("fixture", fixture) print("slots", len(rows)) print("soc_wh", args.soc_wh) print("neg_buy slots", len(neg_buy), "first", neg_buy[0]["interval_start"] if neg_buy else None) print("neg_sell slots", len(neg_sell), "first", neg_sell[0]["interval_start"] if neg_sell else None) print("relax steps", list(SOLVER_RELAX_STEPS)) print() cases: list[tuple[str, dict]] = [ ("permissive masks, 1-pass", dict(permissive_masks=True, two_pass=False)), ("permissive masks, 2-pass", dict(permissive_masks=True, two_pass=True)), ("realistic masks, 1-pass auto-retry", dict(permissive_masks=False, two_pass=False)), ("realistic masks, 2-pass auto-retry", dict(permissive_masks=False, two_pass=True)), ("realistic + row masks from fixture", dict(use_row_masks=True, two_pass=False)), ("strict only (no auto retry)", dict( permissive_masks=False, relaxed_expensive_import=False, relaxed_neg_buy_charge=False, relaxed_neg_prep_hold_only=False, relaxed_neg_prep_window=False, neg_sell_phases_fallback=False, )), ("+ relaxed_expensive_import", dict( permissive_masks=False, relaxed_expensive_import=True, )), ("+ relaxed_neg_buy_charge", dict( permissive_masks=False, relaxed_expensive_import=True, relaxed_neg_buy_charge=True, )), ("+ relaxed_neg_prep_hold_only (evening push kept)", dict( permissive_masks=False, relaxed_expensive_import=True, relaxed_neg_buy_charge=True, relaxed_neg_prep_hold_only=True, )), ("+ relaxed_neg_prep_window (full prep relax)", dict( permissive_masks=False, relaxed_expensive_import=True, relaxed_neg_buy_charge=True, relaxed_neg_prep_hold_only=True, relaxed_neg_prep_window=True, )), ("+ neg_sell_phases_fallback", dict( permissive_masks=False, relaxed_expensive_import=True, relaxed_neg_buy_charge=True, relaxed_neg_prep_hold_only=True, relaxed_neg_prep_window=True, neg_sell_phases_fallback=True, )), ] for label, kw in cases: permissive = kw.pop("permissive_masks", False) use_row_masks = kw.pop("use_row_masks", False) slots = load_slots(rows, permissive_masks=permissive, use_row_masks=use_row_masks) msg, _inp = try_solve(label, slots, args.soc_wh, **kw) print(msg) if __name__ == "__main__": main()