Spaces:

SolarWine
/

api

Sleeping

api / scripts /shadow_mode_test.py

Deploy: 2026 sensor migration + redesign + bucket B endpoints

13fc29d verified 6 months ago

11.4 kB

#!/usr/bin/env python3
"""
Shadow mode validation: run the full upgraded pipeline over a historical week.
Validates three properties:
  1. Sensor errors caught by LLM-generated filters
  2. Baseline predictor (FvCB+ML hybrid) generates valid day-ahead A profiles
  3. Routing agent switches to ML during high-stress and FvCB during cool conditions
Usage
-----
    python scripts/shadow_mode_test.py
    python scripts/shadow_mode_test.py --start 2025-07-20 --end 2025-07-26 -v
"""

from __future__ import annotations

import argparse
import logging
import math
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from config.settings import (
    DP_SLOT_DURATION_MIN,
    SEMILLON_TRANSITION_TEMP_C,
    SIMULATION_LOG_DIR,
)

logger = logging.getLogger("shadow_mode")


# ---------------------------------------------------------------------------
# Test 1: LLM Data Cleaning Filters
# ---------------------------------------------------------------------------

def test_data_cleaning_filters() -> tuple[bool, str]:
    """Verify that LLM-generated data filters catch sensor anomalies."""
    try:
        from src.llm_data_engineer import LLMDataEngineer
        import pandas as pd
        import numpy as np

        # Create synthetic data with known anomalies
        n = 100
        rng = np.random.default_rng(42)
        df = pd.DataFrame({
            "PAR_Den_Avg": np.clip(rng.normal(800, 200, n), 0, 2000),
            "VPD_kPa": np.clip(rng.normal(1.5, 0.5, n), 0, 5),
            "CO2_ppm": np.clip(rng.normal(400, 20, n), 350, 500),
        })

        # Inject anomalies
        df.loc[5, "PAR_Den_Avg"] = -50.0      # negative PAR
        df.loc[10, "PAR_Den_Avg"] = 5000.0     # impossibly high PAR
        df.loc[20, "VPD_kPa"] = -1.0           # negative VPD
        df.loc[30, "CO2_ppm"] = 10.0            # impossibly low CO2

        engineer = LLMDataEngineer()
        cleaned = engineer.apply_cleaning(df)

        # Check that anomalies were handled
        issues = []
        if cleaned.loc[5, "PAR_Den_Avg"] < 0:
            issues.append("negative PAR not clipped")
        if cleaned.loc[10, "PAR_Den_Avg"] > 2500:
            issues.append("impossibly high PAR not clipped")
        if cleaned.loc[20, "VPD_kPa"] < 0:
            issues.append("negative VPD not clipped")

        if issues:
            return False, f"Filter gaps: {'; '.join(issues)}"

        removed = (df.isna().sum().sum() != cleaned.isna().sum().sum()) or \
                  (df.values != cleaned.values).any()
        return True, f"Data cleaning active: {removed}"

    except ImportError as exc:
        return True, f"LLM data engineer not available (OK in test): {exc}"
    except Exception as exc:
        return False, f"Data cleaning failed: {exc}"


# ---------------------------------------------------------------------------
# Test 2: Baseline Predictor Validation
# ---------------------------------------------------------------------------

def test_baseline_predictor() -> tuple[bool, str]:
    """Verify BaselinePredictor generates valid day-ahead A profiles."""
    from src.baseline_predictor import BaselinePredictor

    predictor = BaselinePredictor()

    # Hot July day in Sde Boker
    temps = []
    ghis = []
    for slot in range(96):
        hour = slot * 0.25
        t = 25.0 + 13.0 * max(0, math.sin(math.pi * (hour - 5) / 14)) \
            if 5 <= hour <= 19 else 25.0
        temps.append(t)
        g = max(0, 950 * math.sin(math.pi * (hour - 4) / 12)) \
            if 4 <= hour <= 16 else 0.0
        ghis.append(g)

    predictions = predictor.predict_day(temps, ghis)
    issues = []

    if len(predictions) != 96:
        issues.append(f"expected 96 slots, got {len(predictions)}")

    # Should have non-zero A during daylight
    daylight_a = [predictions[i] for i in range(96) if ghis[i] > 100]
    if not daylight_a or max(daylight_a) == 0:
        issues.append("all daylight A predictions are zero")

    # A should be in reasonable range (0-40 µmol/m²/s)
    if max(predictions) > 50:
        issues.append(f"max A={max(predictions):.1f} > 50 (unrealistic)")

    # Nighttime should be zero
    night_a = [predictions[i] for i in range(96) if ghis[i] < 10]
    non_zero_night = sum(1 for a in night_a if a > 0)
    if non_zero_night > 0:
        issues.append(f"{non_zero_night} non-zero nighttime predictions")

    # A should peak during morning-midday (not late afternoon)
    # Note: at Sde Boker in July, midday temps >35°C cause Rubisco limitation
    # so peak A is biologically correct in the cooler morning hours
    morning_midday = predictions[24:48]  # 6:00-12:00 UTC (local ~9:00-15:00)
    late_afternoon = predictions[56:64]  # 14:00-16:00 UTC
    if morning_midday and late_afternoon and all(a == 0 for a in morning_midday):
        issues.append("all morning-midday A predictions are zero")

    if issues:
        return False, "; ".join(issues)

    peak_a = max(predictions)
    peak_slot = predictions.index(peak_a)
    daylight_mean = sum(daylight_a) / len(daylight_a) if daylight_a else 0
    return True, (
        f"Valid: peak A={peak_a:.1f} at slot {peak_slot} "
        f"({peak_slot // 4:02d}:{(peak_slot % 4) * 15:02d}), "
        f"daylight mean={daylight_mean:.1f}"
    )


# ---------------------------------------------------------------------------
# Test 3: Routing Agent Behaviour
# ---------------------------------------------------------------------------

def test_routing_agent() -> tuple[bool, str]:
    """Verify routing agent switches between FvCB and ML correctly."""
    from src.chatbot.routing_agent import RoutingAgent

    agent = RoutingAgent()
    issues = []

    # Cool morning → should route to FvCB
    cool = agent.route({
        "temp_c": 22.0, "ghi_w_m2": 350.0, "cwsi": 0.15, "vpd": 0.8, "hour": 8,
    })
    if cool != "fvcb":
        issues.append(f"cool morning routed to {cool}, expected fvcb")

    # Hot afternoon, high stress → should route to ML
    hot = agent.route({
        "temp_c": 38.0, "ghi_w_m2": 950.0, "cwsi": 0.72, "vpd": 3.5, "hour": 14,
    })
    if hot != "ml":
        issues.append(f"hot afternoon routed to {hot}, expected ml")

    # Moderate conditions → transition zone (fvcb or ml both acceptable)
    moderate = agent.route({
        "temp_c": 29.5, "ghi_w_m2": 680.0, "cwsi": 0.35, "vpd": 1.8, "hour": 11,
    })
    if moderate not in ("fvcb", "ml"):
        issues.append(f"moderate conditions routed to {moderate}, expected fvcb or ml")

    # Very high VPD → ML
    high_vpd = agent.route({
        "temp_c": 28.0, "ghi_w_m2": 500.0, "cwsi": 0.2, "vpd": 3.0, "hour": 12,
    })
    if high_vpd != "ml":
        issues.append(f"high VPD routed to {high_vpd}, expected ml")

    # Night → FvCB (no stress)
    night = agent.route({
        "temp_c": 18.0, "ghi_w_m2": 0.0, "hour": 22,
    })
    if night != "fvcb":
        issues.append(f"night routed to {night}, expected fvcb")

    if issues:
        return False, "; ".join(issues)

    return True, f"Routing correct: cool→{cool}, hot→{hot}, moderate→{moderate}, vpd→{high_vpd}, night→{night}"


# ---------------------------------------------------------------------------
# Test 4: Full pipeline shadow mode simulation
# ---------------------------------------------------------------------------

def test_shadow_mode_simulation(
    start: date, end: date,
) -> tuple[bool, str]:
    """Run the control loop in shadow mode over a date range."""
    from src.control_loop import ControlLoop

    loop = ControlLoop(dry_run=True)
    issues = []
    total_ticks = 0
    errors = 0
    routes = {"fvcb": 0, "ml": 0, "": 0}
    overrides = 0

    current = start
    while current <= end:
        plan = loop.load_plan(current)
        if plan is None:
            logger.warning("No plan for %s — skipping", current)
            current += timedelta(days=1)
            continue

        for slot_idx in range(96):
            hour = slot_idx // 4
            minute = (slot_idx % 4) * DP_SLOT_DURATION_MIN
            ts = datetime(
                current.year, current.month, current.day,
                hour, minute, 0, tzinfo=timezone.utc,
            )

            try:
                result = loop.tick(timestamp=ts)
                total_ticks += 1

                route = result.model_route or ""
                routes[route] = routes.get(route, 0) + 1

                if result.live_override:
                    overrides += 1

            except Exception as exc:
                errors += 1
                if errors <= 5:
                    logger.error("Tick error at %s: %s", ts, exc)

        current += timedelta(days=1)

    if total_ticks == 0:
        issues.append("no ticks executed")
    if errors > total_ticks * 0.1:
        issues.append(f"{errors} errors out of {total_ticks} ticks (>{10}%)")

    # Should have a mix of FvCB and ML routes during daytime
    # (nighttime all goes to fvcb)
    fvcb_pct = routes.get("fvcb", 0) / max(total_ticks, 1) * 100
    ml_pct = routes.get("ml", 0) / max(total_ticks, 1) * 100

    if issues:
        return False, "; ".join(issues)

    return True, (
        f"Shadow mode OK: {total_ticks} ticks, {errors} errors, "
        f"routes: fvcb={fvcb_pct:.0f}% ml={ml_pct:.0f}%, "
        f"overrides={overrides}"
    )


# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Shadow mode validation of the full agrivoltaic pipeline."
    )
    parser.add_argument(
        "--start", type=str, default="2025-07-01",
        help="Start date for shadow simulation",
    )
    parser.add_argument(
        "--end", type=str, default="2025-07-07",
        help="End date for shadow simulation",
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true",
        help="Enable debug logging",
    )
    args = parser.parse_args()

    level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s %(name)-15s %(levelname)-7s %(message)s",
        datefmt="%H:%M:%S",
    )

    start = date.fromisoformat(args.start)
    end = date.fromisoformat(args.end)

    tests = [
        ("1. Data Cleaning Filters", test_data_cleaning_filters),
        ("2. Baseline Predictor", test_baseline_predictor),
        ("3. Routing Agent", test_routing_agent),
        ("4. Shadow Mode Simulation", lambda: test_shadow_mode_simulation(start, end)),
    ]

    print(f"\nShadow Mode Validation: {start} → {end}")
    print("=" * 60)

    all_passed = True
    for name, test_fn in tests:
        print(f"\n{name}...")
        try:
            passed, msg = test_fn()
            status = "PASS" if passed else "FAIL"
            print(f"  [{status}] {msg}")
            if not passed:
                all_passed = False
        except Exception as exc:
            print(f"  [ERROR] {exc}")
            all_passed = False

    print(f"\n{'=' * 60}")
    print(f"Overall: {'ALL PASSED' if all_passed else 'SOME FAILED'}")
    print(f"{'=' * 60}\n")

    sys.exit(0 if all_passed else 1)


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""
	Shadow mode validation: run the full upgraded pipeline over a historical week.

	Validates three properties:
	1. Sensor errors caught by LLM-generated filters
	2. Baseline predictor (FvCB+ML hybrid) generates valid day-ahead A profiles
	3. Routing agent switches to ML during high-stress and FvCB during cool conditions

	Usage
	-----
	python scripts/shadow_mode_test.py
	python scripts/shadow_mode_test.py --start 2025-07-20 --end 2025-07-26 -v
	"""

	from __future__ import annotations

	import argparse
	import logging
	import math
	import sys
	from datetime import date, datetime, timedelta, timezone
	from pathlib import Path

	sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

	from config.settings import (
	DP_SLOT_DURATION_MIN,
	SEMILLON_TRANSITION_TEMP_C,
	SIMULATION_LOG_DIR,
	)

	logger = logging.getLogger("shadow_mode")


	# ---------------------------------------------------------------------------
	# Test 1: LLM Data Cleaning Filters
	# ---------------------------------------------------------------------------

	def test_data_cleaning_filters() -> tuple[bool, str]:
	"""Verify that LLM-generated data filters catch sensor anomalies."""
	try:
	from src.llm_data_engineer import LLMDataEngineer
	import pandas as pd
	import numpy as np

	# Create synthetic data with known anomalies
	n = 100
	rng = np.random.default_rng(42)
	df = pd.DataFrame({
	"PAR_Den_Avg": np.clip(rng.normal(800, 200, n), 0, 2000),
	"VPD_kPa": np.clip(rng.normal(1.5, 0.5, n), 0, 5),
	"CO2_ppm": np.clip(rng.normal(400, 20, n), 350, 500),
	})

	# Inject anomalies
	df.loc[5, "PAR_Den_Avg"] = -50.0 # negative PAR
	df.loc[10, "PAR_Den_Avg"] = 5000.0 # impossibly high PAR
	df.loc[20, "VPD_kPa"] = -1.0 # negative VPD
	df.loc[30, "CO2_ppm"] = 10.0 # impossibly low CO2

	engineer = LLMDataEngineer()
	cleaned = engineer.apply_cleaning(df)

	# Check that anomalies were handled
	issues = []
	if cleaned.loc[5, "PAR_Den_Avg"] < 0:
	issues.append("negative PAR not clipped")
	if cleaned.loc[10, "PAR_Den_Avg"] > 2500:
	issues.append("impossibly high PAR not clipped")
	if cleaned.loc[20, "VPD_kPa"] < 0:
	issues.append("negative VPD not clipped")

	if issues:
	return False, f"Filter gaps: {'; '.join(issues)}"

	removed = (df.isna().sum().sum() != cleaned.isna().sum().sum()) or \
	(df.values != cleaned.values).any()
	return True, f"Data cleaning active: {removed}"

	except ImportError as exc:
	return True, f"LLM data engineer not available (OK in test): {exc}"
	except Exception as exc:
	return False, f"Data cleaning failed: {exc}"


	# ---------------------------------------------------------------------------
	# Test 2: Baseline Predictor Validation
	# ---------------------------------------------------------------------------

	def test_baseline_predictor() -> tuple[bool, str]:
	"""Verify BaselinePredictor generates valid day-ahead A profiles."""
	from src.baseline_predictor import BaselinePredictor

	predictor = BaselinePredictor()

	# Hot July day in Sde Boker
	temps = []
	ghis = []
	for slot in range(96):
	hour = slot * 0.25
	t = 25.0 + 13.0 * max(0, math.sin(math.pi * (hour - 5) / 14)) \
	if 5 <= hour <= 19 else 25.0
	temps.append(t)
	g = max(0, 950 * math.sin(math.pi * (hour - 4) / 12)) \
	if 4 <= hour <= 16 else 0.0
	ghis.append(g)

	predictions = predictor.predict_day(temps, ghis)
	issues = []

	if len(predictions) != 96:
	issues.append(f"expected 96 slots, got {len(predictions)}")

	# Should have non-zero A during daylight
	daylight_a = [predictions[i] for i in range(96) if ghis[i] > 100]
	if not daylight_a or max(daylight_a) == 0:
	issues.append("all daylight A predictions are zero")

	# A should be in reasonable range (0-40 µmol/m²/s)
	if max(predictions) > 50:
	issues.append(f"max A={max(predictions):.1f} > 50 (unrealistic)")

	# Nighttime should be zero
	night_a = [predictions[i] for i in range(96) if ghis[i] < 10]
	non_zero_night = sum(1 for a in night_a if a > 0)
	if non_zero_night > 0:
	issues.append(f"{non_zero_night} non-zero nighttime predictions")

	# A should peak during morning-midday (not late afternoon)
	# Note: at Sde Boker in July, midday temps >35°C cause Rubisco limitation
	# so peak A is biologically correct in the cooler morning hours
	morning_midday = predictions[24:48] # 6:00-12:00 UTC (local ~9:00-15:00)
	late_afternoon = predictions[56:64] # 14:00-16:00 UTC
	if morning_midday and late_afternoon and all(a == 0 for a in morning_midday):
	issues.append("all morning-midday A predictions are zero")

	if issues:
	return False, "; ".join(issues)

	peak_a = max(predictions)
	peak_slot = predictions.index(peak_a)
	daylight_mean = sum(daylight_a) / len(daylight_a) if daylight_a else 0
	return True, (
	f"Valid: peak A={peak_a:.1f} at slot {peak_slot} "
	f"({peak_slot // 4:02d}:{(peak_slot % 4) * 15:02d}), "
	f"daylight mean={daylight_mean:.1f}"
	)


	# ---------------------------------------------------------------------------
	# Test 3: Routing Agent Behaviour
	# ---------------------------------------------------------------------------

	def test_routing_agent() -> tuple[bool, str]:
	"""Verify routing agent switches between FvCB and ML correctly."""
	from src.chatbot.routing_agent import RoutingAgent

	agent = RoutingAgent()
	issues = []

	# Cool morning → should route to FvCB
	cool = agent.route({
	"temp_c": 22.0, "ghi_w_m2": 350.0, "cwsi": 0.15, "vpd": 0.8, "hour": 8,
	})
	if cool != "fvcb":
	issues.append(f"cool morning routed to {cool}, expected fvcb")

	# Hot afternoon, high stress → should route to ML
	hot = agent.route({
	"temp_c": 38.0, "ghi_w_m2": 950.0, "cwsi": 0.72, "vpd": 3.5, "hour": 14,
	})
	if hot != "ml":
	issues.append(f"hot afternoon routed to {hot}, expected ml")

	# Moderate conditions → transition zone (fvcb or ml both acceptable)
	moderate = agent.route({
	"temp_c": 29.5, "ghi_w_m2": 680.0, "cwsi": 0.35, "vpd": 1.8, "hour": 11,
	})
	if moderate not in ("fvcb", "ml"):
	issues.append(f"moderate conditions routed to {moderate}, expected fvcb or ml")

	# Very high VPD → ML
	high_vpd = agent.route({
	"temp_c": 28.0, "ghi_w_m2": 500.0, "cwsi": 0.2, "vpd": 3.0, "hour": 12,
	})
	if high_vpd != "ml":
	issues.append(f"high VPD routed to {high_vpd}, expected ml")

	# Night → FvCB (no stress)
	night = agent.route({
	"temp_c": 18.0, "ghi_w_m2": 0.0, "hour": 22,
	})
	if night != "fvcb":
	issues.append(f"night routed to {night}, expected fvcb")

	if issues:
	return False, "; ".join(issues)

	return True, f"Routing correct: cool→{cool}, hot→{hot}, moderate→{moderate}, vpd→{high_vpd}, night→{night}"


	# ---------------------------------------------------------------------------
	# Test 4: Full pipeline shadow mode simulation
	# ---------------------------------------------------------------------------

	def test_shadow_mode_simulation(
	start: date, end: date,
	) -> tuple[bool, str]:
	"""Run the control loop in shadow mode over a date range."""
	from src.control_loop import ControlLoop

	loop = ControlLoop(dry_run=True)
	issues = []
	total_ticks = 0
	errors = 0
	routes = {"fvcb": 0, "ml": 0, "": 0}
	overrides = 0

	current = start
	while current <= end:
	plan = loop.load_plan(current)
	if plan is None:
	logger.warning("No plan for %s — skipping", current)
	current += timedelta(days=1)
	continue

	for slot_idx in range(96):
	hour = slot_idx // 4
	minute = (slot_idx % 4) * DP_SLOT_DURATION_MIN
	ts = datetime(
	current.year, current.month, current.day,
	hour, minute, 0, tzinfo=timezone.utc,
	)

	try:
	result = loop.tick(timestamp=ts)
	total_ticks += 1

	route = result.model_route or ""
	routes[route] = routes.get(route, 0) + 1

	if result.live_override:
	overrides += 1

	except Exception as exc:
	errors += 1
	if errors <= 5:
	logger.error("Tick error at %s: %s", ts, exc)

	current += timedelta(days=1)

	if total_ticks == 0:
	issues.append("no ticks executed")
	if errors > total_ticks * 0.1:
	issues.append(f"{errors} errors out of {total_ticks} ticks (>{10}%)")

	# Should have a mix of FvCB and ML routes during daytime
	# (nighttime all goes to fvcb)
	fvcb_pct = routes.get("fvcb", 0) / max(total_ticks, 1) * 100
	ml_pct = routes.get("ml", 0) / max(total_ticks, 1) * 100

	if issues:
	return False, "; ".join(issues)

	return True, (
	f"Shadow mode OK: {total_ticks} ticks, {errors} errors, "
	f"routes: fvcb={fvcb_pct:.0f}% ml={ml_pct:.0f}%, "
	f"overrides={overrides}"
	)


	# ---------------------------------------------------------------------------
	# Runner
	# ---------------------------------------------------------------------------

	def main():
	parser = argparse.ArgumentParser(
	description="Shadow mode validation of the full agrivoltaic pipeline."
	)
	parser.add_argument(
	"--start", type=str, default="2025-07-01",
	help="Start date for shadow simulation",
	)
	parser.add_argument(
	"--end", type=str, default="2025-07-07",
	help="End date for shadow simulation",
	)
	parser.add_argument(
	"--verbose", "-v", action="store_true",
	help="Enable debug logging",
	)
	args = parser.parse_args()

	level = logging.DEBUG if args.verbose else logging.INFO
	logging.basicConfig(
	level=level,
	format="%(asctime)s %(name)-15s %(levelname)-7s %(message)s",
	datefmt="%H:%M:%S",
	)

	start = date.fromisoformat(args.start)
	end = date.fromisoformat(args.end)

	tests = [
	("1. Data Cleaning Filters", test_data_cleaning_filters),
	("2. Baseline Predictor", test_baseline_predictor),
	("3. Routing Agent", test_routing_agent),
	("4. Shadow Mode Simulation", lambda: test_shadow_mode_simulation(start, end)),
	]

	print(f"\nShadow Mode Validation: {start} → {end}")
	print("=" * 60)

	all_passed = True
	for name, test_fn in tests:
	print(f"\n{name}...")
	try:
	passed, msg = test_fn()
	status = "PASS" if passed else "FAIL"
	print(f" [{status}] {msg}")
	if not passed:
	all_passed = False
	except Exception as exc:
	print(f" [ERROR] {exc}")
	all_passed = False

	print(f"\n{'=' * 60}")
	print(f"Overall: {'ALL PASSED' if all_passed else 'SOME FAILED'}")
	print(f"{'=' * 60}\n")

	sys.exit(0 if all_passed else 1)


	if __name__ == "__main__":
	main()