🚧 Add environment maneuverability
This commit is contained in:
Jonas Bengtsson
@@ -0,0 +1,184 @@
|
||||
"""EnvRunner: web-UI-controllable ticker for a plant environment.
|
||||
|
||||
Owns the wall-clock-paced tick task that used to live as a closure in
|
||||
the CLI's ``_serve`` (FR-8.2.17 runner (a)). Moved here so the web UI
|
||||
can control the lifecycle at runtime (spec FR-10.2.15):
|
||||
|
||||
start(ref) load (+ hot reload, FR-10.2.17) + tick
|
||||
stop() cancel the ticker; the field freezes
|
||||
pause() cancel the ticker, keep the instance
|
||||
resume() reschedule the ticker
|
||||
reset() env.reset() — deterministic fresh episode
|
||||
restart() stop + start of the same ref, fresh instance
|
||||
|
||||
State machine: none -> running <-> paused -> done; stop/discard -> none.
|
||||
Exactly one env at a time (one decision source, FR-8.2.15): start is
|
||||
refused while one is loaded.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .scenario.engine import ScenarioError
|
||||
from .scenario.plantenv import PlantEnv, load_env
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .modules import VirtualNode
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EnvRunner:
|
||||
"""Paced ticker + lifecycle for one PlantEnv on a virtual node."""
|
||||
|
||||
def __init__(self, node: VirtualNode) -> None:
|
||||
self.node = node
|
||||
self.env: PlantEnv | None = None
|
||||
self.env_ref: str | None = None
|
||||
self._task: asyncio.Task | None = None
|
||||
self._state: str = "none" # none | running | paused | done
|
||||
self._done_reason: str | None = None
|
||||
|
||||
# -- introspection ---------------------------------------------------------
|
||||
|
||||
@property
|
||||
def state(self) -> str:
|
||||
return self._state
|
||||
|
||||
def status(self) -> dict[str, object]:
|
||||
"""Status snapshot for the web UI (FR-10.2.15)."""
|
||||
return {
|
||||
"ref": self.env_ref,
|
||||
"state": self._state,
|
||||
"reason": self._done_reason,
|
||||
"sim_time_ms": self.env.sim_time_ms if self.env else None,
|
||||
"tick_ms": self.env.TICK_MS if self.env else None,
|
||||
}
|
||||
|
||||
# -- lifecycle --------------------------------------------------------------
|
||||
|
||||
def start(self, ref: str) -> None:
|
||||
"""Load + start an environment (refused while one is loaded).
|
||||
|
||||
``reload=True`` passes through to ``load_env`` (FR-10.2.17):
|
||||
a cached module is invalidated first, so starting after an
|
||||
edit of the source file runs the NEW code.
|
||||
"""
|
||||
if self.env is not None:
|
||||
raise ScenarioError(
|
||||
"an environment is already loaded "
|
||||
f"({self.env_ref!r}); stop it first")
|
||||
env = load_env(ref, node=self.node, reload=True)
|
||||
self.env = env
|
||||
self.env_ref = ref
|
||||
self._state = "running"
|
||||
self._done_reason = None
|
||||
self._task = asyncio.create_task(self._run())
|
||||
log.info("env started: %s", ref)
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Cancel the ticker and drop the env (field freezes)."""
|
||||
await self._cancel_task()
|
||||
self.env = None
|
||||
self.env_ref = None
|
||||
self._state = "none"
|
||||
self._done_reason = None
|
||||
log.info("env stopped")
|
||||
|
||||
async def pause(self) -> None:
|
||||
"""Cancel the ticker, keep the instance (plant stuck fault)."""
|
||||
if self.env is None or self._state not in ("running", "done"):
|
||||
raise ScenarioError(f"cannot pause: state is {self._state!r}")
|
||||
if self._state == "done":
|
||||
raise ScenarioError("environment already done; restart it")
|
||||
await self._cancel_task()
|
||||
self._state = "paused"
|
||||
log.info("env paused (sim %.0f ms)", self.env.sim_time_ms)
|
||||
|
||||
async def resume(self) -> None:
|
||||
"""Reschedule the ticker after a pause."""
|
||||
if self.env is None or self._state != "paused":
|
||||
raise ScenarioError(f"cannot resume: state is {self._state!r}")
|
||||
self._state = "running"
|
||||
self._task = asyncio.create_task(self._run())
|
||||
log.info("env resumed (sim %.0f ms)", self.env.sim_time_ms)
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Deterministic fresh episode (NFR-8.7.3); ticker keeps running."""
|
||||
if self.env is None:
|
||||
raise ScenarioError("no environment loaded")
|
||||
self.env.reset()
|
||||
self._done_reason = None
|
||||
if self._state == "done":
|
||||
# the criterion may hold again on the fresh episode — but
|
||||
# is_done() is re-checked every tick anyway, so just resume
|
||||
self._state = "running"
|
||||
self._task = asyncio.create_task(self._run())
|
||||
log.info("env reset")
|
||||
|
||||
async def restart(self) -> None:
|
||||
"""Fresh instance of the same ref (picks up edited source)."""
|
||||
if self.env_ref is None:
|
||||
raise ScenarioError("no environment to restart")
|
||||
ref = self.env_ref
|
||||
await self.stop()
|
||||
self.start(ref)
|
||||
|
||||
async def discard(self) -> None:
|
||||
"""Silent unload (module-list rebuild, shutdown path)."""
|
||||
await self._cancel_task()
|
||||
self.env = None
|
||||
self.env_ref = None
|
||||
self._state = "none"
|
||||
self._done_reason = None
|
||||
|
||||
def dispose(self) -> None:
|
||||
"""Synchronous unload without awaiting (rebuild path).
|
||||
|
||||
The API handlers ``await discard()`` BEFORE rebuilding; this
|
||||
is the belt-and-braces guarantee for sync callers
|
||||
(``rebuild_node``) that the ticker can never tick the new
|
||||
image: ``cancel()`` takes effect before the loop resumes, and
|
||||
a fully synchronous function yields no loop slots.
|
||||
"""
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
self._task = None
|
||||
self.env = None
|
||||
self.env_ref = None
|
||||
self._state = "none"
|
||||
self._done_reason = None
|
||||
|
||||
# -- ticker -------------------------------------------------------------------
|
||||
|
||||
async def _run(self) -> None:
|
||||
"""Monotonic-scheduled paced ticker (no drift), as in _serve."""
|
||||
assert self.env is not None
|
||||
period = self.env.TICK_MS / 1000.0
|
||||
next_at = time.monotonic()
|
||||
while True:
|
||||
self.env.tick()
|
||||
if self.env.is_done(self.env.observe()):
|
||||
self._state = "done"
|
||||
self._done_reason = "is_done()"
|
||||
log.info("env is_done() satisfied; ticker stopped")
|
||||
return
|
||||
next_at += period
|
||||
delay = next_at - time.monotonic()
|
||||
if delay > 0:
|
||||
await asyncio.sleep(delay)
|
||||
else:
|
||||
next_at = time.monotonic() # catch up, no drift burst
|
||||
|
||||
async def _cancel_task(self) -> None:
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._task = None
|
||||
Reference in New Issue
Block a user