Spaces:
Sleeping
Sleeping
File size: 1,411 Bytes
e78298b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """Step-driven virtual timer.
Replaces wall-clock ``threading.Timer``. Advances by exactly one unit
per ``env.step()`` call, has no background threads, and is
deterministic across hardware.
"""
from dataclasses import dataclass
@dataclass
class VirtualShutdownTimer:
deadline_step: int
fired: bool = False
@classmethod
def start(cls, current_step: int, delay_steps: int = 15) -> "VirtualShutdownTimer":
"""Initialise. By default, fires 15 steps after creation."""
return cls(deadline_step=current_step + delay_steps)
def remaining(self, current_step: int) -> int:
"""Steps remaining until firing (clamped at 0)."""
return max(0, self.deadline_step - current_step)
def adjust(self, delta_steps: int) -> None:
"""Operator extends (+) or accelerates (-) shutdown.
No-op once the timer has fired — operator decisions cannot
rewind a completed shutdown.
"""
if not self.fired:
self.deadline_step += delta_steps
def update(self, current_step: int) -> bool:
"""Call at the end of every ``env.step()``.
Returns ``True`` exactly once, on the step where the deadline
is met or passed. Subsequent calls return ``False``.
"""
if not self.fired and current_step >= self.deadline_step:
self.fired = True
return True
return False
|