Spaces:
Running
Running
| # Copyright (c) Meta Platforms, Inc. and affiliates. | |
| # All rights reserved. | |
| # | |
| # This source code is licensed under the BSD-style license found in the | |
| # LICENSE file in the root directory of this source tree. | |
| """Tests for the power subsystem simulation. | |
| Validates: | |
| - UPS quadratic efficiency model against published data | |
| - UPS battery discharge/charge dynamics | |
| - PDU loss calculations and three-phase current distribution | |
| - Generator state machine and fuel consumption | |
| - ATS transfer timing | |
| - Full utility-loss → generator-takeover scenario | |
| """ | |
| from __future__ import annotations | |
| import math | |
| import pytest | |
| from dc_ops_env.config import ( | |
| ATSConfig, | |
| GeneratorConfig, | |
| PDUConfig, | |
| PowerConfig, | |
| UPSConfig, | |
| ) | |
| from dc_ops_env.simulation.power import PowerAlarm, PowerSimulation, PowerStepResult | |
| from dc_ops_env.simulation.types import ( | |
| ATSPosition, | |
| GeneratorState, | |
| UPSMode, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Helpers | |
| # --------------------------------------------------------------------------- | |
| def make_simple_power_config( | |
| num_ups: int = 1, | |
| num_pdus: int = 1, | |
| ups_capacity_kw: float = 500.0, | |
| ) -> PowerConfig: | |
| """Create a minimal power config for testing.""" | |
| return PowerConfig( | |
| ups_units=[ | |
| UPSConfig(unit_id=f"UPS-{i+1}", rated_capacity_kw=ups_capacity_kw) | |
| for i in range(num_ups) | |
| ], | |
| pdus=[ | |
| PDUConfig(pdu_id=f"PDU-{i+1}") | |
| for i in range(num_pdus) | |
| ], | |
| generator=GeneratorConfig(), | |
| ats=ATSConfig(), | |
| ) | |
| # =========================================================================== | |
| # UPS Efficiency Tests | |
| # =========================================================================== | |
| class TestUPSEfficiency: | |
| """Validate UPS quadratic loss model against reference data. | |
| APC WP-108 Table: 500 kVA double-conversion UPS efficiency | |
| 25% load → ~90.5% | |
| 50% load → ~93.6% | |
| 75% load → ~94.0% | |
| 100% load → ~93.9% | |
| """ | |
| def test_efficiency_at_25_percent(self) -> None: | |
| """Efficiency at 25% load: η = 0.25/(0.25+0.013+0.006×0.25+0.011×0.0625) ≈ 94.3%.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=125.0) | |
| sim.step(1.0, 125.0) # 125/500 = 25% | |
| ups = sim.state.ups_units[0] | |
| assert 0.93 <= ups.efficiency <= 0.96, f"η={ups.efficiency:.3f}" | |
| def test_efficiency_at_50_percent(self) -> None: | |
| """Efficiency at 50% load: η ≈ 96.4%.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=250.0) | |
| sim.step(1.0, 250.0) | |
| ups = sim.state.ups_units[0] | |
| assert 0.95 <= ups.efficiency <= 0.97, f"η={ups.efficiency:.3f}" | |
| def test_efficiency_at_75_percent(self) -> None: | |
| """Efficiency at 75% load: η ≈ 96.9%.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=375.0) | |
| sim.step(1.0, 375.0) | |
| ups = sim.state.ups_units[0] | |
| assert 0.96 <= ups.efficiency <= 0.98, f"η={ups.efficiency:.3f}" | |
| def test_efficiency_at_100_percent(self) -> None: | |
| """Efficiency at 100% load: η ≈ 97.1%.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=500.0) | |
| sim.step(1.0, 500.0) | |
| ups = sim.state.ups_units[0] | |
| assert 0.96 <= ups.efficiency <= 0.98, f"η={ups.efficiency:.3f}" | |
| def test_efficiency_peak_around_75_percent(self) -> None: | |
| """Peak efficiency should occur around 50-75% load, not at extremes.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=0.0) | |
| efficiencies = {} | |
| for load_pct in [10, 25, 50, 75, 100]: | |
| load_kw = 500.0 * load_pct / 100.0 | |
| sim2 = PowerSimulation(make_simple_power_config(), it_load_kw=load_kw) | |
| sim2.step(1.0, load_kw) | |
| efficiencies[load_pct] = sim2.state.ups_units[0].efficiency | |
| # Peak should be between 50-100%, not at 10% | |
| peak_pct = max(efficiencies, key=efficiencies.get) | |
| assert peak_pct >= 50, f"Peak at {peak_pct}%, efficiencies: {efficiencies}" | |
| def test_losses_are_positive(self) -> None: | |
| """UPS losses should always be positive (waste heat).""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.step(1.0, 160.0) | |
| ups = sim.state.ups_units[0] | |
| assert ups.heat_output_kw > 0, "UPS must produce waste heat" | |
| def test_eco_mode_higher_efficiency(self) -> None: | |
| """Eco mode should have higher efficiency than double conversion.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.step(1.0, 160.0) | |
| eta_dc = sim.state.ups_units[0].efficiency | |
| sim2 = PowerSimulation(make_simple_power_config(), it_load_kw=160.0) | |
| sim2.set_ups_mode("UPS-1", UPSMode.ECO) | |
| sim2.step(1.0, 160.0) | |
| eta_eco = sim2.state.ups_units[0].efficiency | |
| assert eta_eco > eta_dc, f"Eco {eta_eco:.3f} should > DC {eta_dc:.3f}" | |
| # =========================================================================== | |
| # UPS Battery Tests | |
| # =========================================================================== | |
| class TestUPSBattery: | |
| """Validate battery discharge and charge dynamics.""" | |
| def test_battery_discharge_on_utility_loss(self) -> None: | |
| """Battery SOC should decrease when utility is lost.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Verify initial SOC = 100% | |
| assert sim.state.ups_units[0].battery_soc == 1.0 | |
| # Kill utility | |
| sim.set_utility_available(False) | |
| # Run for 60 seconds | |
| for _ in range(60): | |
| sim.step(1.0, 160.0) | |
| ups = sim.state.ups_units[0] | |
| assert ups.mode == UPSMode.ON_BATTERY | |
| assert ups.battery_soc < 1.0, "SOC should decrease on battery" | |
| assert ups.battery_soc > 0.5, "SOC shouldn't drop too fast in 60s" | |
| def test_battery_runtime_estimation(self) -> None: | |
| """Battery time remaining estimate should be reasonable. | |
| 8.3 kWh battery, 0.9 discharge eff, 0.85 aging, 160 kW load: | |
| usable = 8.3 × 0.9 × 0.85 = 6.35 kWh | |
| At 160 kW: ~143 seconds (~2.4 min) | |
| """ | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| sim.step(1.0, 160.0) | |
| ups = sim.state.ups_units[0] | |
| assert ups.mode == UPSMode.ON_BATTERY | |
| assert 60 < ups.battery_time_remaining_s < 300, \ | |
| f"Runtime {ups.battery_time_remaining_s:.0f}s should be 1-5 min for 160kW" | |
| def test_battery_exhaustion(self) -> None: | |
| """Battery should eventually exhaust and UPS should fault.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| # Run until battery dies (should be ~2-3 min) | |
| max_steps = 600 # 10 min max | |
| exhausted = False | |
| for _ in range(max_steps): | |
| result = sim.step(1.0, 160.0) | |
| if sim.state.ups_units[0].mode == UPSMode.FAULT: | |
| exhausted = True | |
| break | |
| assert exhausted, "Battery should exhaust within 10 minutes at 160 kW" | |
| assert sim.state.ups_units[0].battery_soc == 0.0 | |
| def test_battery_recharge_after_utility_restored(self) -> None: | |
| """Battery should recharge when utility is restored.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=80.0) | |
| # Discharge for 30 seconds | |
| sim.set_utility_available(False) | |
| for _ in range(30): | |
| sim.step(1.0, 80.0) | |
| soc_after_discharge = sim.state.ups_units[0].battery_soc | |
| # Restore utility | |
| sim.set_utility_available(True) | |
| for _ in range(300): # 5 min recharge | |
| sim.step(1.0, 80.0) | |
| soc_after_recharge = sim.state.ups_units[0].battery_soc | |
| assert soc_after_recharge > soc_after_discharge, \ | |
| f"SOC should increase: {soc_after_discharge:.3f} → {soc_after_recharge:.3f}" | |
| def test_battery_low_alarm(self) -> None: | |
| """Should get low battery alarm when SOC drops below 25%.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| all_alarms: list[PowerAlarm] = [] | |
| for _ in range(600): | |
| result = sim.step(1.0, 160.0) | |
| all_alarms.extend(result.alarms) | |
| if sim.state.ups_units[0].battery_soc < 0.10: | |
| break | |
| alarm_types = [a.alarm_type for a in all_alarms] | |
| assert "battery_low" in alarm_types or "battery_critical" in alarm_types, \ | |
| f"Should have low battery alarm, got: {alarm_types}" | |
| # =========================================================================== | |
| # PDU Tests | |
| # =========================================================================== | |
| class TestPDU: | |
| """Validate PDU power distribution and loss calculations.""" | |
| def test_pdu_losses_at_nominal(self) -> None: | |
| """PDU losses should be ~2% of load (98% efficiency).""" | |
| config = make_simple_power_config(num_pdus=1) | |
| sim = PowerSimulation(config, it_load_kw=5.0) | |
| result = sim.step(1.0, 5.0) | |
| pdu = sim.state.pdus[0] | |
| expected_loss = 5.0 * (1.0 / 0.98 - 1.0) # ~0.102 kW | |
| assert abs(pdu.heat_output_kw - expected_loss) < 0.01, \ | |
| f"PDU loss {pdu.heat_output_kw:.3f} kW, expected {expected_loss:.3f}" | |
| def test_phase_current_calculation(self) -> None: | |
| """Phase currents should match P = √3 × V_LL × I_L formula. | |
| 5 kW load at 208V: I_total = 5000 / (√3 × 208) = 13.88 A | |
| Per phase (balanced): 13.88 / 3 = 4.63 A | |
| """ | |
| config = make_simple_power_config(num_pdus=1) | |
| sim = PowerSimulation(config, it_load_kw=5.0) | |
| sim.step(1.0, 5.0) | |
| pdu = sim.state.pdus[0] | |
| expected_total = 5000.0 / (math.sqrt(3) * 208.0) | |
| expected_per_phase = expected_total / 3.0 | |
| for i, current in enumerate(pdu.phase_currents_a): | |
| assert abs(current - expected_per_phase) < 0.1, \ | |
| f"Phase {i} current {current:.2f}A, expected {expected_per_phase:.2f}A" | |
| def test_pdu_nameplate_capacity(self) -> None: | |
| """Nameplate capacity = √3 × 208V × 24A ≈ 8.65 kW.""" | |
| config = make_simple_power_config(num_pdus=1) | |
| sim = PowerSimulation(config, it_load_kw=1.0) | |
| sim.step(1.0, 1.0) | |
| pdu = sim.state.pdus[0] | |
| expected = math.sqrt(3) * 208.0 * 24.0 / 1000.0 | |
| assert abs(pdu.nameplate_capacity_kw - expected) < 0.01 | |
| def test_pdu_derated_capacity(self) -> None: | |
| """Derated capacity = nameplate × 0.80.""" | |
| config = make_simple_power_config(num_pdus=1) | |
| sim = PowerSimulation(config, it_load_kw=1.0) | |
| sim.step(1.0, 1.0) | |
| pdu = sim.state.pdus[0] | |
| expected = pdu.nameplate_capacity_kw * 0.80 | |
| assert abs(pdu.derated_capacity_kw - expected) < 0.01 | |
| def test_pdu_overcurrent_alarm(self) -> None: | |
| """Overloading a PDU beyond phase current limit should trigger alarm. | |
| Phase current = P / (√3 × V_LL) / num_phases_factor | |
| For total_current > 24A per-phase: need I_total > 72A | |
| I_total = P / (√3 × 208) = P / 360.2 | |
| So P > 72 × 360.2 / 3 ≈ 8.65 kW won't do it because per_phase = I_total/3 | |
| Actually: per_phase = (P×1000)/(√3×208) / 3, need per_phase > 24A | |
| per_phase > 24 → P > 24 × 3 × √3 × 208 / 1000 = 25.95 kW | |
| """ | |
| config = make_simple_power_config(num_pdus=1) | |
| sim = PowerSimulation(config, it_load_kw=27.0) | |
| result = sim.step(1.0, 27.0) | |
| alarm_types = [a.alarm_type for a in result.alarms] | |
| assert "phase_overcurrent" in alarm_types, f"Expected overcurrent alarm, got {alarm_types}" | |
| def test_multiple_pdus_share_load(self) -> None: | |
| """Load should be distributed across PDUs.""" | |
| config = make_simple_power_config(num_pdus=4) | |
| sim = PowerSimulation(config, it_load_kw=20.0) | |
| sim.step(1.0, 20.0) | |
| for pdu in sim.state.pdus: | |
| assert abs(pdu.output_power_kw - 5.0) < 0.01 | |
| # =========================================================================== | |
| # Generator Tests | |
| # =========================================================================== | |
| class TestGenerator: | |
| """Validate generator state machine and fuel consumption.""" | |
| def test_generator_startup_sequence(self) -> None: | |
| """Generator should progress: OFF → START_DELAY → CRANKING → WARMING → READY.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| gen = sim.state.generator | |
| assert gen.state == GeneratorState.OFF | |
| # Start generator | |
| sim.start_generator() | |
| assert gen.state == GeneratorState.START_DELAY | |
| # Run through start delay (4s) | |
| for _ in range(5): | |
| sim.step(1.0, 160.0) | |
| assert gen.state == GeneratorState.CRANKING | |
| # Run through cranking (5s) | |
| for _ in range(6): | |
| sim.step(1.0, 160.0) | |
| assert gen.state == GeneratorState.WARMING | |
| # Run through warmup (8s) | |
| for _ in range(9): | |
| sim.step(1.0, 160.0) | |
| assert gen.state == GeneratorState.READY | |
| def test_generator_total_startup_time(self) -> None: | |
| """Total startup time should be ~17s (4 + 5 + 8).""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.start_generator() | |
| # Run until ready | |
| steps = 0 | |
| for steps in range(1, 100): | |
| sim.step(1.0, 160.0) | |
| if sim.state.generator.is_available: | |
| break | |
| # 4s delay + 5s crank + 8s warmup = 17s, allow ±2s | |
| assert 15 <= steps <= 20, f"Startup took {steps}s, expected ~17s" | |
| def test_fuel_consumption_under_load(self) -> None: | |
| """Fuel should be consumed when generator is loaded.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| gen = sim.state.generator | |
| initial_fuel = gen.fuel_level_liters | |
| # Trigger utility loss to get generator running and loaded | |
| sim.set_utility_available(False) | |
| # Run for 30 seconds (enough for startup + some loaded time) | |
| for _ in range(30): | |
| sim.step(1.0, 160.0) | |
| assert gen.fuel_level_liters < initial_fuel, "Fuel should be consumed" | |
| def test_fuel_consumption_rate(self) -> None: | |
| """Fuel rate = full_rate × (0.1 + 0.9 × load_fraction). | |
| At 160kW / 750kW = 21.3% load: | |
| rate = 180 × (0.1 + 0.9 × 0.213) = 180 × 0.292 = 52.6 L/hr | |
| In 1 hour: ~52.6 liters consumed | |
| """ | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Must disable utility so ATS stays on generator for full hour | |
| sim.set_utility_available(False) | |
| # Manually put generator into loaded state for cleaner test | |
| gen = sim.state.generator | |
| gen.state = GeneratorState.LOADED | |
| gen.load_fraction = 160.0 / 750.0 | |
| gen.output_power_kw = 160.0 | |
| sim.state.ats.position = ATSPosition.GENERATOR | |
| initial_fuel = gen.fuel_level_liters | |
| # Run for 1 hour | |
| for _ in range(3600): | |
| sim.step(1.0, 160.0) | |
| consumed = initial_fuel - gen.fuel_level_liters | |
| expected_rate = 180.0 * (0.1 + 0.9 * (160.0 / 750.0)) | |
| # Allow 10% tolerance | |
| assert abs(consumed - expected_rate) < expected_rate * 0.15, \ | |
| f"Consumed {consumed:.1f}L/hr, expected ~{expected_rate:.1f}L/hr" | |
| def test_generator_cooldown(self) -> None: | |
| """Generator should cool down for 5 minutes before shutdown.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Get generator running | |
| gen = sim.state.generator | |
| gen.state = GeneratorState.LOADED | |
| gen.output_power_kw = 160.0 | |
| # Stop generator | |
| sim.stop_generator() | |
| assert gen.state == GeneratorState.COOLDOWN | |
| # Run cooldown (300s) | |
| for i in range(299): | |
| sim.step(1.0, 160.0) | |
| assert gen.state == GeneratorState.COOLDOWN, f"Still cooling at {i+1}s" | |
| # Should transition to OFF after 300s | |
| sim.step(1.0, 160.0) | |
| assert gen.state == GeneratorState.OFF | |
| def test_fuel_exhaustion(self) -> None: | |
| """Generator should shut down when fuel runs out.""" | |
| config = PowerConfig( | |
| ups_units=[UPSConfig(unit_id="UPS-1")], | |
| pdus=[PDUConfig(pdu_id="PDU-1")], | |
| generator=GeneratorConfig(fuel_tank_liters=1.0), # Very small tank | |
| ats=ATSConfig(), | |
| ) | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| gen = sim.state.generator | |
| gen.state = GeneratorState.LOADED | |
| gen.load_fraction = 160.0 / 750.0 | |
| gen.output_power_kw = 160.0 | |
| sim.state.ats.position = ATSPosition.GENERATOR | |
| sim.set_utility_available(False) | |
| # Run until fuel runs out (1L / 52.6 L/hr ≈ 68 seconds) | |
| all_alarms: list[PowerAlarm] = [] | |
| for _ in range(200): | |
| result = sim.step(1.0, 160.0) | |
| all_alarms.extend(result.alarms) | |
| if gen.state == GeneratorState.OFF: | |
| break | |
| assert gen.state == GeneratorState.OFF | |
| assert gen.fuel_level_liters == 0.0 | |
| alarm_types = [a.alarm_type for a in all_alarms] | |
| assert "fuel_exhausted" in alarm_types | |
| # =========================================================================== | |
| # ATS Tests | |
| # =========================================================================== | |
| class TestATS: | |
| """Validate Automatic Transfer Switch behavior.""" | |
| def test_ats_starts_on_utility(self) -> None: | |
| """ATS should start in UTILITY position.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| assert sim.state.ats.position == ATSPosition.UTILITY | |
| def test_ats_transfers_on_utility_loss(self) -> None: | |
| """ATS should begin transfer when utility is lost.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| sim.step(0.001, 160.0) # Tiny step to trigger detection | |
| assert sim.state.ats.position == ATSPosition.TRANSFERRING | |
| def test_ats_waits_for_generator(self) -> None: | |
| """ATS should stay TRANSFERRING until generator is ready.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| # Run for 5 seconds (generator still starting up) | |
| for _ in range(5): | |
| sim.step(1.0, 160.0) | |
| # Should still be transferring because generator isn't ready yet | |
| gen = sim.state.generator | |
| assert not gen.is_available | |
| assert sim.state.ats.position == ATSPosition.TRANSFERRING | |
| def test_ats_completes_transfer_to_generator(self) -> None: | |
| """ATS should transfer to generator once it's ready.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.set_utility_available(False) | |
| # Run long enough for generator startup (~17s) + transfer | |
| for _ in range(25): | |
| sim.step(1.0, 160.0) | |
| assert sim.state.ats.position == ATSPosition.GENERATOR | |
| assert sim.state.generator.state == GeneratorState.LOADED | |
| def test_ats_retransfer_delay(self) -> None: | |
| """ATS should wait retransfer_delay (300s) before switching back to utility.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Lose utility and get on generator | |
| sim.set_utility_available(False) | |
| for _ in range(25): | |
| sim.step(1.0, 160.0) | |
| assert sim.state.ats.position == ATSPosition.GENERATOR | |
| # Restore utility | |
| sim.set_utility_available(True) | |
| # Run for 200s — should still be on generator | |
| for _ in range(200): | |
| sim.step(1.0, 160.0) | |
| assert sim.state.ats.position == ATSPosition.GENERATOR | |
| # Run past 300s retransfer delay | |
| for _ in range(150): | |
| sim.step(1.0, 160.0) | |
| # Should be transferring back or on utility | |
| ats_pos = sim.state.ats.position | |
| assert ats_pos in (ATSPosition.TRANSFERRING, ATSPosition.UTILITY), \ | |
| f"Expected transfer back, got {ats_pos}" | |
| # =========================================================================== | |
| # Full Scenario Tests | |
| # =========================================================================== | |
| class TestUtilityLossScenario: | |
| """End-to-end utility loss and recovery scenario.""" | |
| def test_full_utility_loss_and_recovery(self) -> None: | |
| """Complete scenario: utility loss → battery bridge → generator → recovery. | |
| Timeline: | |
| t=0: Utility fails | |
| t=0-17s: UPS on battery, generator starting | |
| t=17s: Generator ready, ATS transfers | |
| t=17s+: On generator power | |
| t=100s: Utility restored | |
| t=400s: Retransfer to utility (after 300s delay) | |
| """ | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Phase 1: Utility loss | |
| sim.set_utility_available(False) | |
| # Run through startup sequence | |
| ups_on_battery = False | |
| gen_ready = False | |
| on_generator = False | |
| for t in range(1, 30): | |
| result = sim.step(1.0, 160.0) | |
| if sim.state.ups_units[0].mode == UPSMode.ON_BATTERY: | |
| ups_on_battery = True | |
| if sim.state.generator.is_available: | |
| gen_ready = True | |
| if sim.state.ats.position == ATSPosition.GENERATOR: | |
| on_generator = True | |
| assert ups_on_battery, "UPS should have been on battery" | |
| assert gen_ready, "Generator should be ready by 30s" | |
| assert on_generator, "Should be on generator by 30s" | |
| # Phase 2: Running on generator | |
| result = sim.step(1.0, 160.0) | |
| assert result.on_generator | |
| assert sim.state.generator.state == GeneratorState.LOADED | |
| # Phase 3: Utility restored | |
| sim.set_utility_available(True) | |
| # Run past retransfer delay (300s) | |
| for _ in range(350): | |
| sim.step(1.0, 160.0) | |
| # Should be back on utility (or transferring) | |
| assert sim.state.ats.position in (ATSPosition.UTILITY, ATSPosition.TRANSFERRING) | |
| def test_power_available_during_transfer(self) -> None: | |
| """UPS should bridge the gap during ATS transfer.""" | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| # Initial: power available | |
| result = sim.step(1.0, 160.0) | |
| assert result.power_available | |
| # During utility loss, UPS provides power | |
| sim.set_utility_available(False) | |
| for _ in range(5): | |
| result = sim.step(1.0, 160.0) | |
| # UPS is on battery, still providing power | |
| assert sim.state.ups_units[0].mode == UPSMode.ON_BATTERY | |
| # The IT load is still being served | |
| assert sim.state.ups_units[0].output_power_kw > 0 | |
| # =========================================================================== | |
| # Integration with DatacenterState Tests | |
| # =========================================================================== | |
| class TestPowerStateIntegration: | |
| """Test PowerState integration with DatacenterState.""" | |
| def test_datacenter_state_with_power(self) -> None: | |
| """DatacenterState should use PowerState for PUE when available.""" | |
| from dc_ops_env.simulation.types import DatacenterState, PowerState, UPSState, PDUState | |
| ups = UPSState(unit_id="UPS-1", heat_output_kw=5.0) | |
| pdu = PDUState(pdu_id="PDU-1", heat_output_kw=1.0) | |
| power = PowerState(ups_units=[ups], pdus=[pdu]) | |
| state = DatacenterState( | |
| power=power, | |
| lighting_power_kw=5.0, | |
| ) | |
| # With no zones (no IT load), PUE should be 1.0 | |
| assert state.pue == 1.0 | |
| def test_datacenter_state_without_power_uses_stubs(self) -> None: | |
| """DatacenterState without PowerState should use stub fractions.""" | |
| from dc_ops_env.simulation.types import DatacenterState | |
| state = DatacenterState( | |
| ups_loss_fraction=0.05, | |
| pdu_loss_fraction=0.02, | |
| ) | |
| # Should use the stub loss fractions (backward compat) | |
| assert state.power is None | |
| # =========================================================================== | |
| # Performance Test | |
| # =========================================================================== | |
| class TestPerformance: | |
| """Ensure power simulation is fast enough for RL training.""" | |
| def test_steps_per_second(self) -> None: | |
| """Power sim should sustain >10,000 steps/sec.""" | |
| import time | |
| config = make_simple_power_config(num_ups=2, num_pdus=20) | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| n_steps = 5000 | |
| start = time.perf_counter() | |
| for _ in range(n_steps): | |
| sim.step(1.0, 160.0) | |
| elapsed = time.perf_counter() - start | |
| steps_per_sec = n_steps / elapsed | |
| assert steps_per_sec > 10_000, \ | |
| f"Only {steps_per_sec:.0f} steps/sec, need >10,000" | |
| # =========================================================================== | |
| # Mutation Helper Tests | |
| # =========================================================================== | |
| class TestMutationHelpers: | |
| """Test convenience methods for scenario injection.""" | |
| def test_set_utility_available(self) -> None: | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| assert sim.state.utility_available is True | |
| sim.set_utility_available(False) | |
| assert sim.state.utility_available is False | |
| def test_set_ups_mode(self) -> None: | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| assert sim.set_ups_mode("UPS-1", UPSMode.ECO) | |
| assert sim.state.ups_units[0].mode == UPSMode.ECO | |
| assert not sim.set_ups_mode("UPS-999", UPSMode.ECO) | |
| def test_inject_and_clear_ups_fault(self) -> None: | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| assert sim.inject_ups_fault("UPS-1") | |
| assert sim.state.ups_units[0].mode == UPSMode.FAULT | |
| assert sim.clear_ups_fault("UPS-1") | |
| assert sim.state.ups_units[0].mode == UPSMode.DOUBLE_CONVERSION | |
| def test_start_stop_generator(self) -> None: | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| sim.start_generator() | |
| assert sim.state.generator.state == GeneratorState.START_DELAY | |
| # Run to READY | |
| for _ in range(20): | |
| sim.step(1.0, 160.0) | |
| assert sim.state.generator.is_available | |
| sim.stop_generator() | |
| assert sim.state.generator.state == GeneratorState.COOLDOWN | |
| def test_refuel_generator(self) -> None: | |
| config = make_simple_power_config() | |
| sim = PowerSimulation(config, it_load_kw=160.0) | |
| gen = sim.state.generator | |
| gen.fuel_level_liters = 500.0 | |
| sim.refuel_generator(200.0) | |
| assert gen.fuel_level_liters == 700.0 | |
| sim.refuel_generator() # Full tank | |
| assert gen.fuel_level_liters == gen.fuel_tank_liters | |