Spaces:

SolarWine
/

api

Sleeping

api / src /control_loop.py
Vectorize Farquhar, DI ControlLoop, gate pipeline, budget audit, chatbot Hebrew
15be6bb verified 8 months ago
30.6 kB
"""
ControlLoop: the 15-minute agrivoltaic control cycle.
Each tick:
  1. Fetch live sensor data (IMS weather + TB vine sensors)
  2. Load/validate the day-ahead plan for today
  3. Look up the planned offset for the current slot
  4. Run live gate check (may override plan if conditions diverged)
  5. Check energy budget (block intervention if budget exhausted)
  6. Run CommandArbiter (priority stack + hysteresis)
  7. Resolve per-tracker fleet overrides (rare; default = all same angle)
  8. Dispatch angle to trackers via TrackerDispatcher
  9. Spend energy budget for the slot
 10. Check plan divergence and trigger re-plan if needed
 11. Log the result
The loop can run as:
  - **one-shot**: ``loop.tick()`` — execute one cycle (called externally)
  - **continuous**: ``loop.run()`` — blocking loop with 15-min sleep
  - **plan-only**: ``loop.tick(dry_run=True)`` — compute decisions without sending
"""

from __future__ import annotations

import json
import logging
import time
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional

import pandas as pd

from config.settings import (
    ANGLE_TOLERANCE_DEG,
    DAILY_PLAN_PATH,
    DP_SLOT_DURATION_MIN,
    PLAN_DIVERGENCE_THRESHOLD_KWH,
    PLAN_DIVERGENCE_THRESHOLD_SLOTS,
    PLAN_REPLAN_COOLDOWN_SLOTS,
    SIMULATION_LOG_PATH,
)

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Tick result
# ---------------------------------------------------------------------------

@dataclass
class TickResult:
    """Output of a single control loop tick."""

    timestamp: datetime
    slot_index: int                     # 0–95
    stage_id: str = "unknown"

    # Plan lookup
    plan_offset_deg: float = 0.0        # what the day-ahead plan says
    plan_gate_passed: bool = False

    # Live override
    live_gate_passed: bool = False
    live_override: bool = False         # True if live data diverged from plan
    override_reason: Optional[str] = None

    # Arbiter decision
    target_angle: float = 0.0
    dispatch: bool = False
    source: str = ""

    # Dispatch result
    trackers_verified: int = 0
    trackers_total: int = 0
    dispatch_error: Optional[str] = None

    # Energy cost
    energy_cost_kwh: float = 0.0        # energy sacrificed by this slot's offset

    # Budget tracking
    budget_spent_kwh: float = 0.0       # actual amount deducted from budget
    budget_remaining_kwh: float = 0.0   # daily budget remaining after this slot

    # Model routing
    model_route: str = ""               # "fvcb" or "ml" — which model was selected

    # Fleet overrides (per-tracker angles, if any differ from default)
    fleet_overrides: Optional[Dict[str, float]] = None

    # Plan divergence tracking
    divergence_cumulative_kwh: float = 0.0
    divergence_consecutive: int = 0
    replan_triggered: bool = False

    # Sensor snapshot
    air_temp_c: Optional[float] = None
    ghi_w_m2: Optional[float] = None
    wind_speed_ms: Optional[float] = None

    def to_dict(self) -> dict:
        return {k: (v.isoformat() if isinstance(v, datetime) else v)
                for k, v in self.__dict__.items()}


# ---------------------------------------------------------------------------
# ControlLoop
# ---------------------------------------------------------------------------

class ControlLoop:
    """15-minute agrivoltaic control loop.
    All dependencies are injected via ``__init__()`` with sensible defaults.
    Pass explicit instances for testing (e.g. mock dispatcher).
    Parameters
    ----------
    dry_run : bool
        If True, compute decisions but don't send commands to trackers.
    plan_path : Path
        Path to the day-ahead plan JSON file.
    log_path : Path
        Path for simulation log output.
    arbiter, dispatcher, astro, hub, modes, fleet, budget_planner, router
        Injectable dependencies — pass None (default) to auto-create.
    """

    def __init__(
        self,
        dry_run: bool = True,
        plan_path: Path = DAILY_PLAN_PATH,
        log_path: Path = SIMULATION_LOG_PATH,
        *,
        arbiter=None,
        dispatcher=None,
        astro=None,
        hub=None,
        modes=None,
        fleet=None,
        budget_planner=None,
        router=None,
    ):
        self.dry_run = dry_run
        self.plan_path = plan_path
        self.log_path = log_path

        # Dependencies — lazy-create if not injected
        self._arbiter = arbiter
        self._dispatcher = dispatcher
        self._astro = astro
        self._hub = hub
        self._modes = modes
        self._fleet = fleet
        self._budget_planner = budget_planner
        self._router = router

        self._schedulers: Dict[str, object] = {}
        self._current_plan: Optional[dict] = None
        self._tick_log: List[dict] = []

        # Daily budget state (reset each day)
        self._daily_budget_plan: Optional[dict] = None
        self._daily_budget_date: Optional[date] = None

        # Divergence tracking (reset on re-plan or new day)
        self._divergence_cumulative_kwh: float = 0.0
        self._divergence_consecutive: int = 0
        self._last_replan_slot: int = -99
        self._replan_count: int = 0

    # ------------------------------------------------------------------
    # Lazy component init (auto-create when not injected)
    # ------------------------------------------------------------------

    @property
    def arbiter(self):
        if self._arbiter is None:
            from src.command_arbiter import CommandArbiter
            self._arbiter = CommandArbiter()
        return self._arbiter

    @property
    def dispatcher(self):
        if self._dispatcher is None:
            from src.tracker_dispatcher import TrackerDispatcher
            self._dispatcher = TrackerDispatcher(dry_run=self.dry_run)
        return self._dispatcher

    @property
    def astro(self):
        if self._astro is None:
            from src.command_arbiter import AstronomicalTracker
            self._astro = AstronomicalTracker()
        return self._astro

    @property
    def hub(self):
        if self._hub is None:
            from src.data.data_providers import DataHub
            self._hub = DataHub.default()
        return self._hub

    @property
    def modes(self):
        if self._modes is None:
            from src.operational_modes import OperationalModeChecker
            self._modes = OperationalModeChecker()
        return self._modes

    @property
    def fleet(self):
        if self._fleet is None:
            from src.tracker_fleet import TrackerFleet
            self._fleet = TrackerFleet()
        return self._fleet

    @property
    def budget_planner(self):
        if self._budget_planner is None:
            from src.energy_budget import EnergyBudgetPlanner
            self._budget_planner = EnergyBudgetPlanner()
        return self._budget_planner

    @property
    def router(self):
        if self._router is None:
            from src.chatbot.routing_agent import RoutingAgent
            self._router = RoutingAgent()
        return self._router

    # ------------------------------------------------------------------
    # Plan loading
    # ------------------------------------------------------------------

    def _build_persistence_forecast(self) -> tuple[list[float], list[float]]:
        """Build 96-slot temp/GHI forecast from last available IMS day."""
        ims_df = self.hub.weather.get_dataframe()
        if ims_df.empty:
            return [25.0] * 96, [0.0] * 96

        df = ims_df.copy()
        if "timestamp_utc" in df.columns:
            df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)
            df = df.set_index("timestamp_utc")

        last_day = df.index.max().normalize()
        day_data = df[df.index.normalize() == last_day]
        if len(day_data) < 10:
            last_day -= pd.Timedelta(days=1)
            day_data = df[df.index.normalize() == last_day]

        temps = [25.0] * 96
        ghis = [0.0] * 96
        for _, row in day_data.iterrows():
            slot = row.name.hour * 4 + row.name.minute // 15
            if 0 <= slot < 96:
                t = row.get("air_temperature_c")
                if pd.notna(t):
                    temps[slot] = float(t)
                g = row.get("ghi_w_m2")
                if pd.notna(g):
                    ghis[slot] = float(g)
        return temps, ghis

    def _compute_daily_budget(self, target: date) -> float:
        """Compute the daily energy budget from the annual/monthly hierarchy."""
        annual = self.budget_planner.compute_annual_plan(target.year)
        month_budget = annual["monthly_budgets"].get(target.month, 0.5)
        weekly = self.budget_planner.compute_weekly_plan(target, month_budget)
        dow = target.weekday()
        return weekly["daily_budgets_kWh"][min(dow, 6)]

    def load_plan(self, target_date: Optional[date] = None) -> Optional[dict]:
        """Load the day-ahead plan for the given date."""
        target = target_date or date.today()

        # Try loading from file
        if self.plan_path.exists():
            try:
                with open(self.plan_path) as f:
                    plan = json.load(f)
                if plan.get("target_date") == str(target):
                    self._current_plan = plan
                    logger.info("Loaded plan for %s (%d slots)",
                                target, len(plan.get("slots", [])))
                    return plan
            except Exception as exc:
                logger.warning("Failed to load plan from %s: %s", self.plan_path, exc)

        # No plan file or wrong date — compute on the fly
        try:
            from src.day_ahead_planner import DayAheadPlanner

            temps, ghis = self._build_persistence_forecast()
            daily_budget = self._compute_daily_budget(target)

            planner = DayAheadPlanner()
            plan_obj = planner.plan_day(target, temps, ghis, max(daily_budget, 0.1))
            plan = plan_obj.to_dict()

            # Save for reuse
            self.plan_path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.plan_path, "w") as f:
                json.dump(plan, f, indent=2)

            self._current_plan = plan
            return plan

        except Exception as exc:
            logger.error("Plan generation failed: %s", exc)
            return None

    def _get_slot_plan(self, slot_index: int) -> Optional[dict]:
        """Look up the planned offset for a given slot."""
        if not self._current_plan:
            return None
        slots = self._current_plan.get("slots", [])
        for s in slots:
            t = s.get("time", "")
            try:
                h, m = map(int, t.split(":"))
                s_idx = h * 4 + m // 15
                if s_idx == slot_index:
                    return s
            except (ValueError, AttributeError):
                continue
        return None

    # ------------------------------------------------------------------
    # Energy budget
    # ------------------------------------------------------------------

    def _ensure_daily_budget(self, today: date) -> Optional[dict]:
        """Load or reuse the daily slot-level budget plan."""
        if self._daily_budget_plan and self._daily_budget_date == today:
            return self._daily_budget_plan

        # Try restoring from Redis (survives worker restarts)
        try:
            from src.data.redis_cache import get_redis
            redis = get_redis()
            if redis:
                cached = redis.get_json("control:budget")
                if cached and cached.get("date") == str(today):
                    self._daily_budget_plan = cached["plan"]
                    self._daily_budget_date = today
                    logger.info("Restored daily budget from Redis for %s", today)
                    return self._daily_budget_plan
        except Exception:
            pass

        try:
            daily_budget = self._compute_daily_budget(today)
            self._daily_budget_plan = self.budget_planner.compute_daily_plan(
                today, daily_budget,
            )
            self._daily_budget_date = today

            # Reset divergence tracking for new day
            self._divergence_cumulative_kwh = 0.0
            self._divergence_consecutive = 0
            self._last_replan_slot = -99

            # Persist to Redis
            self._persist_budget(today)

            return self._daily_budget_plan
        except Exception as exc:
            logger.warning("Failed to compute daily budget: %s", exc)
            return None

    def _persist_budget(self, today: date) -> None:
        """Save daily budget state to Redis for cross-process access."""
        try:
            from src.data.redis_cache import get_redis
            import json as _json
            redis = get_redis()
            if redis and self._daily_budget_plan:
                payload = {
                    "date": str(today),
                    "plan": _json.loads(_json.dumps(self._daily_budget_plan, default=str)),
                }
                redis.set_json("control:budget", payload, ttl=86400)
        except Exception as exc:
            logger.debug("Budget Redis persist failed: %s", exc)

    @staticmethod
    def _slot_key(now: datetime) -> str:
        """Format a datetime as a slot key like '10:15'."""
        return f"{now.hour:02d}:{(now.minute // 15) * 15:02d}"

    # ------------------------------------------------------------------
    # Fleet overrides (Task 1)
    # ------------------------------------------------------------------

    def _resolve_fleet_overrides(
        self, now: datetime, theta_astro: float,
    ) -> Dict[str, float]:
        """Resolve per-tracker angle overrides from TrackerFleet assignments.
        Returns an empty dict in the common case (all trackers follow the
        arbiter's angle).  Only returns overrides for trackers that have
        an explicit non-tracking assignment active right now.
        """
        from src.tracker_fleet import tracker_id_to_name
        from src.tracker_scheduler import TrackerScheduler, PLAN_LIBRARY

        overrides: Dict[str, float] = {}
        try:
            best = self.fleet.get_all_best_assignments(now)
        except Exception as exc:
            logger.debug("Fleet assignment lookup skipped: %s", exc)
            return overrides

        for tracker_id, assignment in best.items():
            if assignment is None:
                continue

            plan_id = assignment.plan_id
            # Get or create scheduler for this plan
            if plan_id not in self._schedulers:
                if assignment.plan_file:
                    plan_path = Path(assignment.plan_file)
                    if plan_path.exists():
                        self._schedulers[plan_id] = TrackerScheduler(
                            plan_file=plan_path,
                        )
                    else:
                        logger.warning("Plan file not found: %s", plan_path)
                        continue
                elif plan_id in PLAN_LIBRARY:
                    self._schedulers[plan_id] = TrackerScheduler(
                        plan_data=PLAN_LIBRARY[plan_id],
                    )
                else:
                    logger.debug("Unknown plan_id %r, skipping", plan_id)
                    continue

            sched = self._schedulers[plan_id]
            event = sched.get_event(now)
            if event is None:
                continue

            mode = event.get("mode")
            event_angle = event.get("angle")

            if mode == "tracking" or mode is None:
                # Same as default astronomical tracking — no override needed
                continue
            elif mode == "antiTracking" and event_angle is not None:
                overrides[tracker_id_to_name(tracker_id)] = theta_astro + event_angle
            elif mode == "fixed_angle" and event_angle is not None:
                overrides[tracker_id_to_name(tracker_id)] = event_angle

        return overrides

    # ------------------------------------------------------------------
    # Plan divergence (Task 3)
    # ------------------------------------------------------------------

    def _check_plan_divergence(
        self,
        slot_index: int,
        planned_offset: float,
        actual_offset: float,
        planned_cost: float,
        actual_cost: float,
    ) -> bool:
        """Track divergence between plan and execution. Return True if re-plan needed."""
        cost_diff = abs(planned_cost - actual_cost)
        offset_diverged = abs(planned_offset - actual_offset) > ANGLE_TOLERANCE_DEG

        self._divergence_cumulative_kwh += cost_diff

        if offset_diverged:
            self._divergence_consecutive += 1
        else:
            self._divergence_consecutive = 0

        # Check cooldown
        if slot_index - self._last_replan_slot < PLAN_REPLAN_COOLDOWN_SLOTS:
            return False

        if self._divergence_cumulative_kwh >= PLAN_DIVERGENCE_THRESHOLD_KWH:
            logger.warning(
                "Cumulative divergence %.3f kWh >= %.3f threshold; triggering re-plan",
                self._divergence_cumulative_kwh, PLAN_DIVERGENCE_THRESHOLD_KWH,
            )
            return True

        if self._divergence_consecutive >= PLAN_DIVERGENCE_THRESHOLD_SLOTS:
            logger.warning(
                "%d consecutive divergent slots >= %d threshold; triggering re-plan",
                self._divergence_consecutive, PLAN_DIVERGENCE_THRESHOLD_SLOTS,
            )
            return True

        return False

    def _trigger_replan(self, now: datetime, slot_index: int) -> bool:
        """Re-generate the day-ahead plan from the current slot onward."""
        today = now.date()
        daily_bp = self._ensure_daily_budget(today)
        spent = daily_bp["cumulative_spent"] if daily_bp else 0.0
        remaining = (daily_bp["daily_total_kWh"] - spent) if daily_bp else 0.0

        if remaining <= 0:
            logger.info("Re-plan skipped: no budget remaining")
            return False

        try:
            from src.day_ahead_planner import DayAheadPlanner

            temps, ghis = self._build_persistence_forecast()

            planner = DayAheadPlanner()
            plan_obj = planner.plan_day(today, temps, ghis, max(remaining, 0.01))
            plan = plan_obj.to_dict()

            # Save for reuse
            self.plan_path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.plan_path, "w") as f:
                json.dump(plan, f, indent=2)

            self._current_plan = plan
            self._last_replan_slot = slot_index
            self._divergence_cumulative_kwh = 0.0
            self._divergence_consecutive = 0
            self._replan_count += 1

            n_slots = len(plan.get("slots", []))
            logger.info(
                "Re-plan #%d at slot %d: %d slots, %.4f kWh remaining budget",
                self._replan_count, slot_index, n_slots, remaining,
            )
            return True
        except Exception as exc:
            logger.error("Re-plan failed: %s", exc)
            return False

    # ------------------------------------------------------------------
    # Main tick
    # ------------------------------------------------------------------

    def tick(self, timestamp: Optional[datetime] = None) -> TickResult:
        """Execute one control loop cycle.
        Parameters
        ----------
        timestamp : datetime, optional
            Override current time (for simulation/replay).
        """
        now = timestamp or datetime.now(tz=timezone.utc)
        slot_index = now.hour * 4 + now.minute // 15

        result = TickResult(timestamp=now, slot_index=slot_index)

        # 1. Load plan if needed
        today = now.date() if hasattr(now, 'date') else date.today()
        if (not self._current_plan or
                self._current_plan.get("target_date") != str(today)):
            self.load_plan(today)

        # 2. Fetch live weather
        try:
            wx = self.hub.weather.get_current()
            if "error" not in wx:
                result.air_temp_c = wx.get("air_temperature_c")
                result.ghi_w_m2 = wx.get("ghi_w_m2")
                result.wind_speed_ms = wx.get("wind_speed_ms")
        except Exception as exc:
            logger.warning("Weather fetch failed: %s", exc)

        # 2b. Route model selection (FvCB vs ML) based on live conditions
        try:
            telemetry = {
                "temp_c": result.air_temp_c,
                "ghi_w_m2": result.ghi_w_m2,
                "hour": now.hour,
            }
            result.model_route = self.router.route(telemetry)
        except Exception as exc:
            logger.debug("Model routing failed: %s", exc)
            result.model_route = "fvcb"

        # 3. Get astronomical tracking angle
        theta_astro = self.astro.get_angle(now)

        # 4. Look up plan for this slot
        slot_plan = self._get_slot_plan(slot_index)
        if slot_plan:
            result.plan_offset_deg = slot_plan.get("offset_deg", 0.0)
            result.plan_gate_passed = slot_plan.get("gate_passed", False)
            result.energy_cost_kwh = slot_plan.get("energy_cost_kwh", 0.0)
            result.stage_id = self._current_plan.get("stage_id", "unknown")
        else:
            logger.debug("No plan slot for index %d — defaulting to astronomical", slot_index)

        # 5. Live gate check — override plan if conditions diverged
        # Intentionally simpler than DayAheadPlanner._check_gate():
        # the planner has forecast CWSI + FvCB shading_helps; the live gate
        # only checks real-time temp and GHI as hard constraints.
        planned_offset = result.plan_offset_deg
        live_offset = planned_offset  # default: follow the plan

        if result.air_temp_c is not None:
            from config.settings import (
                NO_SHADE_BEFORE_HOUR,
                SEMILLON_TRANSITION_TEMP_C,
                SHADE_ELIGIBLE_GHI_ABOVE,
            )

            if planned_offset > 0:
                blocked = False
                reason = ""

                if now.hour < NO_SHADE_BEFORE_HOUR:
                    blocked, reason = True, "morning — no shading before 10:00"
                elif result.air_temp_c < SEMILLON_TRANSITION_TEMP_C:
                    blocked, reason = True, f"temp {result.air_temp_c:.0f}°C < {SEMILLON_TRANSITION_TEMP_C:.0f}°C"
                elif result.ghi_w_m2 is not None and result.ghi_w_m2 < SHADE_ELIGIBLE_GHI_ABOVE:
                    blocked, reason = True, f"GHI {result.ghi_w_m2:.0f} < {SHADE_ELIGIBLE_GHI_ABOVE:.0f}"

                if blocked:
                    live_offset = 0.0
                    result.live_override = True
                    result.override_reason = reason
                    logger.info("Live override: plan offset %.0f° → 0° (%s)",
                                planned_offset, reason)

            result.live_gate_passed = live_offset > 0

        # 5b. Budget guard — block intervention if daily budget exhausted
        if live_offset > 0:
            daily_bp = self._ensure_daily_budget(today)
            if daily_bp:
                sk = self._slot_key(now)
                slot_remaining = daily_bp["slot_budgets"].get(sk, 0.0)
                margin_remaining = daily_bp["daily_margin_remaining_kWh"]
                if slot_remaining + margin_remaining <= 0:
                    live_offset = 0.0
                    result.live_override = True
                    result.override_reason = "daily energy budget exhausted"
                    logger.info("Budget guard: forcing astronomical (budget depleted)")

        # 6. Build engine result for arbiter
        target_angle = theta_astro + live_offset
        engine_result = {
            "angle": target_angle,
            "action": f"plan_offset_{live_offset:.0f}deg",
        }

        # Check operational modes (wind stow, heat shield, harvest)
        mode_override = self.modes.check_all(
            wind_speed_ms=result.wind_speed_ms,
            air_temp_c=result.air_temp_c,
            theta_astro=theta_astro,
            current_date=today,
        )
        weather_override = mode_override.to_weather_override() if mode_override else None

        # 7. Arbitrate
        decision = self.arbiter.arbitrate(
            timestamp=now,
            engine_result=engine_result,
            theta_astro=theta_astro,
            weather_override=weather_override,
        )

        result.target_angle = decision.angle
        result.dispatch = decision.dispatch
        result.source = decision.source.value if hasattr(decision.source, 'value') else str(decision.source)

        # 7b. Resolve per-tracker fleet overrides (rare; most ticks return {})
        fleet_overrides = self._resolve_fleet_overrides(now, theta_astro)
        if fleet_overrides:
            result.fleet_overrides = fleet_overrides
            logger.info("Fleet overrides active: %s", fleet_overrides)

        # 8. Dispatch to trackers
        if decision.dispatch:
            try:
                dispatch_result = self.dispatcher.dispatch(
                    decision, angle_overrides=fleet_overrides or None,
                )
                result.trackers_verified = dispatch_result.n_success
                result.trackers_total = len(dispatch_result.trackers)
                if not dispatch_result.all_verified:
                    failed = [t.device_name for t in dispatch_result.trackers if not t.verified]
                    result.dispatch_error = f"failed: {', '.join(failed)}"
            except Exception as exc:
                result.dispatch_error = str(exc)
                logger.error("Dispatch failed: %s", exc)

        # 9. Spend energy budget
        if result.energy_cost_kwh > 0:
            daily_bp = self._ensure_daily_budget(today)
            if daily_bp:
                sk = self._slot_key(now)
                result.budget_spent_kwh = self.budget_planner.spend_slot(
                    daily_bp, sk, result.energy_cost_kwh,
                )
                result.budget_remaining_kwh = (
                    sum(daily_bp["slot_budgets"].values())
                    + daily_bp["daily_margin_remaining_kWh"]
                )
                # Persist updated budget to Redis
                self._persist_budget(today)

                if result.budget_spent_kwh < result.energy_cost_kwh:
                    logger.warning(
                        "Budget shortfall: requested %.4f kWh, spent %.4f kWh (slot %s)",
                        result.energy_cost_kwh, result.budget_spent_kwh, sk,
                    )

        # 10. Check plan divergence and trigger re-plan if needed
        if slot_plan:
            actual_offset = 0.0 if result.live_override else live_offset
            needs_replan = self._check_plan_divergence(
                slot_index=slot_index,
                planned_offset=result.plan_offset_deg,
                actual_offset=actual_offset,
                planned_cost=slot_plan.get("energy_cost_kwh", 0.0),
                actual_cost=result.energy_cost_kwh,
            )
            result.divergence_cumulative_kwh = self._divergence_cumulative_kwh
            result.divergence_consecutive = self._divergence_consecutive
            if needs_replan:
                result.replan_triggered = self._trigger_replan(now, slot_index)

        # 11. Log
        self._tick_log.append(result.to_dict())
        logger.info(
            "Tick %02d:%02d slot=%d angle=%.1f° offset=%.0f° dispatch=%s source=%s"
            " budget_remaining=%.3f kWh%s",
            now.hour, now.minute, slot_index, decision.angle,
            live_offset, decision.dispatch, decision.source,
            result.budget_remaining_kwh,
            f" [OVERRIDE: {result.override_reason}]" if result.live_override else "",
        )

        # 12. Budget audit — append slot to parquet log
        try:
            from src.budget_audit import BudgetAuditLog
            BudgetAuditLog().log_slot(result)
        except Exception as exc:
            logger.debug("Budget audit log skipped: %s", exc)

        return result

    # ------------------------------------------------------------------
    # Continuous run
    # ------------------------------------------------------------------

    def run(self, max_ticks: Optional[int] = None) -> None:
        """Run the control loop continuously (blocking).
        Parameters
        ----------
        max_ticks : int, optional
            Stop after this many ticks (for testing). None = run forever.
        """
        logger.info("Control loop starting (dry_run=%s)", self.dry_run)
        tick_count = 0

        while max_ticks is None or tick_count < max_ticks:
            try:
                result = self.tick()
                tick_count += 1
            except Exception as exc:
                logger.error("Tick failed: %s", exc)

            # Sleep until next 15-min boundary
            now = datetime.now(tz=timezone.utc)
            next_slot = now.replace(
                minute=(now.minute // DP_SLOT_DURATION_MIN + 1) * DP_SLOT_DURATION_MIN % 60,
                second=0, microsecond=0,
            )
            if next_slot <= now:
                next_slot += timedelta(hours=1)
            sleep_sec = (next_slot - now).total_seconds()
            logger.debug("Sleeping %.0f s until %s", sleep_sec, next_slot)
            time.sleep(max(sleep_sec, 1.0))

    # ------------------------------------------------------------------
    # Log access
    # ------------------------------------------------------------------

    def get_log(self) -> List[dict]:
        """Return all tick results from this session."""
        return list(self._tick_log)

    def save_log(self, path: Optional[Path] = None) -> Path:
        """Save tick log to JSON file."""
        out = path or self.log_path.with_suffix(".json")
        out.parent.mkdir(parents=True, exist_ok=True)
        with open(out, "w") as f:
            json.dump(self._tick_log, f, indent=2, default=str)
        logger.info("Saved %d tick results to %s", len(self._tick_log), out)
        return out