shrike/power.py

Part of SHRIKE.

"""Power generation, life support, and the silent-running posture.

The capacitor is the tactical blood supply: weapons, the shield, the
afterburner and the scoop all spend from it through :meth:`PowerSystem.request`
and :meth:`PowerSystem.drain`, and nothing else is allowed to touch the number.
Generation comes from socketed modules, and each one prices its output in a
different currency:

* :class:`SolarWings` are silent and free, but only pay in lit biomes, take two
  seconds to unfold or fold away, and widen the ship's hitbox while out.
* :class:`Generator` pays anywhere and pays best, and charges fuel and noise.
* :class:`RTG` pays a trickle, silently, forever, and is the only source that
  survives silent running.

Silent running is the posture that inverts the whole economy: generation
collapses to the RTG, the generator shuts down, weapons go offline (enforced by
``weapons.py`` off ``SILENT_RUNNING_CHANGED``), and the signature meter starts
falling toward the act floor instead of climbing.

Underneath all of it sits the hull's own emergency bus, a trickle no module
supplies and no posture stops. It is what makes being out of power a setback
rather than the end of the run: it cannot pay for a fight, but it always pays
for the next shot eventually.

Life support and the fuel burn tick here too, so that the ship's breaches and
the generator's appetite are read from one place per frame.
"""

import importlib.util
import math
from collections.abc import Callable, Iterator
from weakref import WeakKeyDictionary, WeakSet

from simvx.core import Input, MeshInstance3D, Node, Node3D, Quat, Signal, Vec3

from . import balance
from .runtime import Services, SignalNames

#: Wings per solar array. Balance sizes one wing; the array is always a pair,
#: so a sheared wing costs exactly half the array's output.
SOLAR_WING_COUNT = 2

#: How often deferred signal wiring and power-source discovery re-walk the
#: tree, in seconds. Sectors, enemies and socketed modules all come and go
#: mid-run, so both passes are periodic rather than one-shot.
REWIRE_INTERVAL_S = 0.5

#: The hull's emergency bus, in energy per second. Always on, whatever is
#: fitted and whatever the posture, and far under any weapon's draw: it buys a
#: way out of a dead capacitor, never a sustained fight.
POWER_EMERGENCY_TRICKLE_PER_S = balance.POWER_EMERGENCY_TRICKLE_PER_S

#: Capacitor level a weapon refuses to fire below. Above it a shot is always
#: allowed, even one the tank cannot fully cover, so a fight never ends because
#: the capacitor was a fraction short of the exact cost.
WEAPONS_ENERGY_FLOOR = balance.WEAPONS_ENERGY_FLOOR

#: Seconds between ``ENERGY_DENIED`` emissions per consumer. A held beam over
#: an empty capacitor asks every frame; one click and one flash a second says
#: "no power" without turning the mix into a geiger counter. Denials are still
#: refused every frame, only the cue is limited.
ENERGY_DENIED_CUE_INTERVAL_S = 1.0

#: What C says on a run that has not unlocked the posture yet. The onboarding
#: schedule holds silent running back until run three; before that the key must
#: not silently do nothing, and it must not act either, because a whole-mix
#: low-pass with no HUD light and no card is a haunted soundtrack, not a
#: mechanic.
SILENT_RUNNING_LOCKED_TOAST = "SILENT RUNNING COMES ONLINE ON A LATER FLIGHT"

# ============================================================================
# The modules' own geometry
#
# The three generation modules move real state (a widened hitbox, thirty hit
# points a wing, a fuel burn, a hum that is the signature warning) and for a
# long time none of them put a single triangle on the hull. A pilot pressed
# X, the array's signal fired, the capacitor started filling, and the ship on
# screen was unchanged; the whole power economy was a HUD reading with no
# object behind it. The constants below are how each module reads at a glance.
# ============================================================================

#: Quarter turn about +Y that brings an art-kit build (nose along +X) onto the
#: hull's own frame, where the nose runs down -Z. The ship applies exactly this
#: to its hull art; a socketed module inherits the hull's transform, so it
#: applies it too rather than inventing a second convention.
ART_QUARTER_TURN = Quat.from_axis_angle(Vec3(0.0, 1.0, 0.0), math.pi * 0.5)

#: How far a folded wing stands out of the flight plane. Just short of upright,
#: so a stowed array still catches an edge of light and reads as folded rather
#: than as missing.
SOLAR_FOLD_RADIANS = math.radians(84.0)
#: Span a fully folded wing still shows, as a fraction of its built span: the
#: panels concertina into the root rather than shrinking to nothing.
SOLAR_STOWED_SPAN = 0.18
#: Cell glow as a multiple of the palette's accent strength, at zero output and
#: at the array's full rated output. The floor is what keeps a deployed array in
#: a nebula visible as unlit hardware rather than absent.
SOLAR_GLOW_DARK = 0.12
SOLAR_GLOW_LIT = 1.4
#: What a wing that has taken damage but still carries cells is dimmed to.
SOLAR_GLOW_DAMAGED = 0.35
#: Hit points below which a wing counts as visibly damaged, as a fraction.
SOLAR_DAMAGED_FRACTION = 0.6

#: The generator's vent pulses at the hum's own rate, between these multiples of
#: the accent strength. Off is off: a stopped generator shows a dark vent, which
#: is the difference the silent-running posture is played around.
GENERATOR_HUM_HZ = 1.7
GENERATOR_GLOW_MIN = 0.55
GENERATOR_GLOW_MAX = 1.6
#: How often the vent is actually rewritten while the generator turns. The hum
#: is under two cycles a second, so thirty steps a second draws the same pulse
#: the eye sees; rewriting it once per frame instead ties the cost of an idle
#: animation to the panel's refresh rate, which is how a game gets hotter on
#: better hardware for no visible gain.
GENERATOR_VENT_REFRESH_S = 1.0 / 30.0

#: The RTG's window never moves. It is the only source silent running leaves
#: running, so a dark hull with one warm band still lit is what the posture
#: looks like from outside.
RTG_GLOW = 0.75


def _artkit():
    """The art kit, or None in a tree built without it.

    Modules are meant to survive a partial checkout the same way the ship does:
    without the kit a generator is still a generator, it simply has no vent to
    light up.
    """
    if importlib.util.find_spec("shrike.artkit") is None:
        return None
    from . import artkit

    return artkit


def _accent_materials(node: Node3D) -> list:
    """Every emissive accent material under *node*, in tree order."""
    from .artkit import ROLE_ACCENT

    return [i.material for i in node.find_all(MeshInstance3D) if i.name == ROLE_ACCENT and i.material is not None]


def accent_strength() -> float:
    """The player palette's accent strength, looked up once.

    Every lit module scales its glow by this, and the palette is a constant of
    the art kit, so the lookup is memoised rather than repeated per module per
    frame.
    """
    global _ACCENT_STRENGTH
    if _ACCENT_STRENGTH is None:
        from .artkit import palette

        _ACCENT_STRENGTH = float(palette("player").accent_strength)
    return _ACCENT_STRENGTH


#: Memo behind :func:`accent_strength`.
_ACCENT_STRENGTH: float | None = None


def is_weapon_consumer(consumer: str) -> bool:
    """True if *consumer* is a weapon id, and so spends under the reserve floor."""
    return consumer in balance.WEAPONS


def iter_tree_nodes(tree) -> Iterator[Node]:
    """Every node in *tree*: the singletons' subtrees first, then the scene."""
    seen: set[int] = set()
    roots = list(tree.singletons.values())
    if tree.root is not None:
        roots.append(tree.root)
    for root in roots:
        for node in root.walk():
            if id(node) not in seen:
                seen.add(id(node))
                yield node


# ============================================================================
# Deferred signal wiring
# ============================================================================


class SignalWiring:
    """Connect an owner's handlers to emitters that need not exist yet.

    Modules mount in an order no single module controls, and the interesting
    emitters (the sector, a refinery, a screamer that spawned nine seconds ago)
    appear and vanish during a run. A one-shot ``connect`` in ``on_ready``
    therefore misses most of them. This walks the tree on an interval and
    connects every emitter carrying a wanted signal exactly once, tracking what
    it has already wired in a weak map so a destroyed node costs nothing.

    The interval leaves one blind spot: a node that emits during its own
    ``on_ready`` has, by definition, never been swept. An emitter about to do
    that calls :meth:`sweep_all` first, which brings every live wiring up to
    date before the signal fires (the sector does this for its one-shot
    ``sector_entered``).

    Usage::

        self._wiring = SignalWiring(self)
        self._wiring.want(SignalNames.VAULT_HACKED, self._on_vault)
        self._wiring.sweep()          # in on_ready, for emitters already up
        self._wiring.poll(dt)         # in on_update, for the ones still coming
    """

    #: Every live wiring (a ``WeakSet[SignalWiring]``), so an emitter can
    #: demand an immediate global sweep.
    _instances: WeakSet = WeakSet()

    def __init__(self, owner: Node, *, interval_s: float = REWIRE_INTERVAL_S):
        self._owner = owner
        self._interval = float(interval_s)
        self._handlers: dict[str, Callable] = {}
        self._wired: WeakKeyDictionary[Node, set[str]] = WeakKeyDictionary()
        self._since_sweep = self._interval
        SignalWiring._instances.add(self)

    @classmethod
    def sweep_all(cls) -> None:
        """Sweep every live wiring now.

        For an emitter mounted moments ago and about to fire a signal no
        interval sweep can have seen it carrying yet. Sweeps are idempotent,
        so the only cost is the tree walks, and the wirings sharing a tree
        share one walk.
        """
        by_tree: dict[int, tuple[object, list[SignalWiring]]] = {}
        for wiring in list(cls._instances):
            tree = wiring._owner.tree
            if tree is None or not wiring._handlers:
                wiring._since_sweep = 0.0
                continue
            by_tree.setdefault(id(tree), (tree, []))[1].append(wiring)
        for tree, wirings in by_tree.values():
            cls._sweep_together(tree, wirings)

    @classmethod
    def _sweep_together(cls, tree, wirings: list[SignalWiring]) -> None:
        """Walk *tree* once and offer every node to each of *wirings*."""
        for wiring in wirings:
            wiring._since_sweep = 0.0
        for node in iter_tree_nodes(tree):
            for wiring in wirings:
                wiring._wire_node(node)

    def want(self, signal_name: str, handler: Callable) -> None:
        """Route every emitter of *signal_name* to *handler*."""
        self._handlers[signal_name] = handler

    def poll(self, dt: float) -> None:
        """Sweep if the interval has elapsed. Cheap to call every frame.

        The walk is shared. A dozen wirings are live in a run, all on the same
        interval, all asking the same two hundred and fifty nodes the same
        question, so the first one whose timer expires walks the tree and
        offers each node to every other wiring that is due this frame. A
        partner is counted as due one frame early, because the ones that have
        not been polled yet are exactly one *dt* behind this one; that is what
        pulls them into lockstep instead of leaving them walking one after
        another. Sweeps are idempotent, so the wiring that results is identical
        to a dozen separate walks; only the walking is done once.
        """
        self._since_sweep += dt
        if self._since_sweep < self._interval:
            return
        tree = self._owner.tree
        if tree is None or not self._handlers:
            self._since_sweep = 0.0
            return
        due = [self]
        due.extend(
            other
            for other in SignalWiring._instances
            if other is not self
            and other._handlers
            and other._since_sweep + dt >= other._interval
            and other._owner.tree is tree
        )
        SignalWiring._sweep_together(tree, due)

    def sweep(self) -> None:
        """Connect to every wanted emitter currently in the tree."""
        self._since_sweep = 0.0
        tree = self._owner.tree
        if tree is None or not self._handlers:
            return
        for node in iter_tree_nodes(tree):
            self._wire_node(node)

    def _wire_node(self, node: Node) -> None:
        """Connect this wiring to whatever *node* emits that it wants."""
        wired = self._wired.get(node)
        if wired is not None and len(wired) == len(self._handlers):
            return
        for name, handler in self._handlers.items():
            if wired is not None and name in wired:
                continue
            signal = getattr(node, name, None)
            if not isinstance(signal, Signal):
                continue
            signal.connect(handler)
            if wired is None:
                wired = set()
                self._wired[node] = wired
            wired.add(name)


# ============================================================================
# Generation modules
# ============================================================================


class PowerSource(Node3D):
    """Base for the socketed modules that feed the capacitor.

    A source is pure declaration: it reports what it produces and what that
    production costs per second, and :class:`PowerSystem` does the ticking. The
    system finds sources by walking the tree, so a module starts paying as soon
    as it is socketed and stops the moment it is removed, with no registration
    handshake to leak.
    """

    #: True if the source keeps producing under silent running. Only the RTG.
    silent_safe: bool = False
    #: Ledger key the signature meter files this source's noise under.
    signature_reason: str = "power"

    def output_per_s(self) -> float:
        """Energy this source adds to the capacitor each second."""
        return 0.0

    def potential_output_per_s(self) -> float:
        """What this source pays once fully brought up, whatever it pays now.

        The ranking of the fit rather than of the frame: a folded wing and an
        idle generator both report what they would pay, which is what decides
        which module the hull treats as its primary source.
        """
        return self.output_per_s()

    def fuel_per_s(self) -> float:
        """Fuel this source burns each second."""
        return 0.0

    def signature_per_s(self) -> float:
        """Signature this source adds each second."""
        return 0.0

    def set_biome(self, biome: balance.BiomeSpec) -> None:
        """Told the sector's biome; sources that care about light override."""

    def shut_down(self) -> None:
        """Stop producing. Called when silent running engages or fuel runs dry."""

    def power_system(self) -> PowerSystem | None:
        """The run's power system, or None before it is registered."""
        tree = self.tree
        if tree is None:
            return None
        return tree.singletons.get(Services.POWER)


class SolarWings(PowerSource):
    """A pair of unfolding solar wings: free power that makes you a bigger target.

    Deployment is a two-second animation in both directions, and the wings pay
    in proportion to how far out they are, so a panicked retraction gives up its
    power gradually rather than at a cliff. Each wing carries its own hit points
    and half the array's output; flak that shears one halves the yield for the
    rest of the sector. Output scales with the biome's light, which is 2x in the
    Solar Shallows and nothing at all in a nebula.

    Wings that are the hull's primary source come out by themselves the moment
    they are fitted, fully and without the animation. Folding them is a real
    decision (a smaller target, no power) and folding them back out is worth two
    seconds of exposure; starting the run folded was neither, only a ship that
    quietly ran flat before its first fight.
    """

    solar_state_changed = Signal(str)
    wing_destroyed = Signal(int)

    #: The four animation states, in the order a full cycle visits them.
    STATES = ("retracted", "deploying", "deployed", "retracting")

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.state = "retracted"
        self.wing_hp = [balance.SOLAR_WING_HP] * SOLAR_WING_COUNT
        self.deployment = 0.0
        self._light_multiplier = 1.0
        self._auto_deploy_pending = True
        #: One pivot node per wing, in ``wing_hp`` order, or empty without art.
        self.wing_nodes: list[Node3D] = []
        self._cell_materials: list = []
        #: The state the model on the hull was last built from. See
        #: :meth:`_visual_key`.
        self._visual_shown: tuple | None = None

    # --- Geometry ----------------------------------------------------------

    def on_ready(self):
        """Hang the array's own geometry and put it in its current state."""
        kit = _artkit()
        if kit is None:
            return
        art = self.add_child(kit.build_solar_wings())
        art.rotation = ART_QUARTER_TURN
        for name in kit.SOLAR_WING_NAMES[:SOLAR_WING_COUNT]:
            wing = next((child for child in art.children if child.name == name), None)
            if wing is None:
                continue
            self.wing_nodes.append(wing)
            materials = _accent_materials(wing)
            self._cell_materials.append(materials[0] if materials else None)
        # The two states the pilot reads off the hull are also the two the
        # module announces, so the announcement is what drives the model: a
        # refresh that only ran per frame would be true by coincidence.
        self.solar_state_changed.connect(self._on_state_shown)
        self.wing_destroyed.connect(self._on_wing_lost)
        self.refresh_visual_state()

    def _on_state_shown(self, state: str) -> None:
        del state
        self.refresh_visual_state()

    def _on_wing_lost(self, index: int) -> None:
        del index
        self.refresh_visual_state()

    def _visual_key(self) -> tuple:
        """Everything the model is a function of, as one comparable value.

        The wings are geometry driven by three numbers, and for most of a run
        none of them move: an array parked deployed in one biome is the same
        pair of panels frame after frame. Rebuilding a quaternion, writing two
        transforms and re-lighting the cells to arrive at the state already on
        the hull is the whole of that cost, so the frame tick asks this first.
        """
        return (round(self.deployment, 4), round(self.glow_fraction(), 4), tuple(self.wing_hp))

    def refresh_visual_state(self) -> None:
        """Put every wing where its deployment, its light and its damage say.

        A destroyed wing leaves the hull outright: half the array's output is
        gone for the rest of the sector, and a panel still hanging there would
        say the opposite. A damaged one stays, dimmed.
        """
        self._visual_shown = self._visual_key()
        if not self.wing_nodes:
            return
        span = SOLAR_STOWED_SPAN + (1.0 - SOLAR_STOWED_SPAN) * self.deployment
        fold = SOLAR_FOLD_RADIANS * (1.0 - self.deployment)
        lit = self.glow_fraction()
        for index, wing in enumerate(self.wing_nodes):
            hp = self.wing_hp[index] if index < len(self.wing_hp) else 0.0
            if hp <= 0.0:
                wing.visible = False
                continue
            wing.visible = True
            # Port and starboard hinge in opposite senses about the shared
            # fore-aft axis, so both tips rise rather than one folding through
            # the hull.
            sense = 1.0 if index % 2 == 0 else -1.0
            wing.rotation = Quat.from_axis_angle(Vec3(1.0, 0.0, 0.0), sense * fold)
            wing.scale = Vec3(1.0, 1.0, span)
            material = self._cell_materials[index] if index < len(self._cell_materials) else None
            if material is None:
                continue
            glow = SOLAR_GLOW_DARK + (SOLAR_GLOW_LIT - SOLAR_GLOW_DARK) * lit
            if hp < balance.SOLAR_WING_HP * SOLAR_DAMAGED_FRACTION:
                glow = min(glow, SOLAR_GLOW_DAMAGED)
            material.emissive_strength = accent_strength() * glow

    def glow_fraction(self) -> float:
        """How hard the cells are working, 0 to 1, against the array's best light.

        Live output over what a whole undamaged array would make in the
        brightest biome, so the glint answers "is this paying?" rather than
        "are the wings out?", and an array unfolded in a nebula stays dark.
        """
        best = balance.SOLAR_OUTPUT_SOLAR_SHALLOWS_PER_S
        if best <= 0.0:
            return 0.0
        return min(1.0, max(0.0, self.output_per_s() / best))

    # --- Deployment --------------------------------------------------------

    def toggle(self) -> None:
        """Reverse the wings: unfold if folded or folding, fold otherwise."""
        self._set_state("retracting" if self.state in ("deployed", "deploying") else "deploying")

    def deploy_now(self) -> None:
        """Snap the wings out with no animation: the state a run starts in."""
        self.deployment = 1.0
        self._set_state("deployed")

    def is_primary_source(self) -> bool:
        """True if nothing else fitted to this hull can out-pay the wings.

        Rated output decides it, so wings bolted on beside a generator stay the
        backup the player unfolds deliberately, and wings in a nebula (which pay
        nothing at all) stay folded rather than widening the hitbox for free.
        """
        mine = self.potential_output_per_s()
        if mine <= 0.0:
            return False
        tree = self.tree
        if tree is None:
            return True
        for node in iter_tree_nodes(tree):
            if node is not self and isinstance(node, PowerSource) and node.potential_output_per_s() > mine:
                return False
        return True

    def hitbox_scale(self) -> float:
        """Multiplier the ship applies to its hull radius while the wings are out."""
        return 1.0 + balance.SOLAR_HITBOX_WIDEN_FRACTION * self.deployment

    def _set_state(self, state: str) -> None:
        if state == self.state:
            return
        self.state = state
        self.solar_state_changed(state)

    # --- Damage ------------------------------------------------------------

    def damage_wing(self, index: int, amount: float) -> bool:
        """Apply *amount* to one wing. Returns True if this shot destroyed it."""
        before = self.wing_hp[index]
        if before <= 0.0:
            return False
        self.wing_hp[index] = max(0.0, before - float(amount))
        if self.wing_hp[index] > 0.0:
            return False
        self.wing_destroyed(index)
        return True

    def live_wings(self) -> int:
        """How many wings still carry cells."""
        return sum(1 for hp in self.wing_hp if hp > 0.0)

    # --- Production --------------------------------------------------------

    def set_biome(self, biome: balance.BiomeSpec) -> None:
        self._light_multiplier = biome.solar_mult

    def potential_output_per_s(self) -> float:
        per_wing = balance.SOLAR_OUTPUT_PER_S * self._light_multiplier / SOLAR_WING_COUNT
        return per_wing * self.live_wings()

    def output_per_s(self) -> float:
        return self.potential_output_per_s() * self.deployment

    # --- Lifecycle ---------------------------------------------------------

    def on_update(self, dt: float):
        if self._auto_deploy_pending:
            # One shot, on the first frame the wings are in a mounted tree: by
            # then every other module has mounted and the sector's light is
            # known, so "am I the primary source" has a real answer.
            self._auto_deploy_pending = False
            if self.state == "retracted" and self.is_primary_source():
                self.deploy_now()
        if Input.is_action_just_pressed("solar_wings"):
            self.toggle()
        if self.state == "deploying":
            self.deployment += dt / balance.SOLAR_DEPLOY_S
            if self.deployment >= 1.0:
                self.deployment = 1.0
                self._set_state("deployed")
        elif self.state == "retracting":
            self.deployment -= dt / balance.SOLAR_RETRACT_S
            if self.deployment <= 0.0:
                self.deployment = 0.0
                self._set_state("retracted")
        # The two-second unfold and the glint that tracks the biome's light are
        # both continuous, so the model is refreshed off the frame as well as on
        # the state change that starts and ends it, but only while one of the
        # three numbers it is built from has actually moved. A settled array,
        # which is what an array is nearly all the time, costs one comparison.
        if self._visual_key() != self._visual_shown:
            self.refresh_visual_state()


class Generator(PowerSource):
    """The loud option: power anywhere, paid for in fuel and in attention.

    Running the generator is the single clearest statement of the game's theme.
    It hums in the mix, and that hum is the signature warning, so flicking it on
    for a fight and killing it to go dark is the posture change every fight is
    built around. It refuses to start under silent running and stops itself the
    moment the tank runs dry.
    """

    generator_changed = Signal(bool)

    signature_reason = "generator"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.running = False
        self._vent = None
        self._hum_phase = 0.0
        #: Seconds until the vent is next rewritten. See
        #: :data:`GENERATOR_VENT_REFRESH_S`.
        self._vent_due = 0.0

    # --- Geometry ----------------------------------------------------------

    def on_ready(self):
        """Hang the block, and light its vent if it is already turning."""
        kit = _artkit()
        if kit is None:
            return
        art = self.add_child(kit.build_generator())
        art.rotation = ART_QUARTER_TURN
        materials = _accent_materials(art)
        self._vent = materials[0] if materials else None
        self.generator_changed.connect(self._on_running_shown)
        self.refresh_visual_state()

    def _on_running_shown(self, running: bool) -> None:
        del running
        self.refresh_visual_state()

    def refresh_visual_state(self) -> None:
        """Light the vent in step with the hum, or leave it dark.

        The pulse is the hum made visible. The generator's loudness is the
        signature warning the whole posture game is played around, and a pilot
        who has muted the mix or cannot hear it had no way at all to tell a
        running generator from a stopped one.
        """
        if self._vent is None:
            return
        if not self.running:
            self._vent.emissive_strength = 0.0
            return
        pulse = 0.5 + 0.5 * math.sin(self._hum_phase * math.tau)
        strength = GENERATOR_GLOW_MIN + (GENERATOR_GLOW_MAX - GENERATOR_GLOW_MIN) * pulse
        self._vent.emissive_strength = accent_strength() * strength

    def toggle(self) -> None:
        """Start or stop the generator. Starting is refused while dark."""
        if not self.running:
            system = self.power_system()
            if system is not None and (system.silent_running or system.fuel <= 0.0):
                return
        self.set_running(not self.running)

    def set_running(self, active: bool) -> None:
        """Set the running state, emitting the hum cue only on a real change."""
        active = bool(active)
        if active == self.running:
            return
        self.running = active
        self.generator_changed(active)

    def shut_down(self) -> None:
        self.set_running(False)

    def output_per_s(self) -> float:
        return balance.GENERATOR_OUTPUT_PER_S if self.running else 0.0

    def potential_output_per_s(self) -> float:
        return balance.GENERATOR_OUTPUT_PER_S

    def fuel_per_s(self) -> float:
        return balance.GENERATOR_FUEL_PER_S if self.running else 0.0

    def signature_per_s(self) -> float:
        return balance.GENERATOR_SIGNATURE_PER_S if self.running else 0.0

    def on_update(self, dt: float):
        if Input.is_action_just_pressed("generator"):
            self.toggle()
        if not self.running:
            # A dark vent is a state, not an animation: the stop wrote it once
            # through ``generator_changed`` and nothing moves it afterwards.
            return
        self._hum_phase += dt * GENERATOR_HUM_HZ
        self._vent_due -= dt
        if self._vent_due <= 0.0:
            self._vent_due = GENERATOR_VENT_REFRESH_S
            self.refresh_visual_state()


class RTG(PowerSource):
    """A radioisotope trickle: small, silent, and the only power left when dark."""

    silent_safe = True

    def on_ready(self):
        """Hang the drum, with its window lit and left alone.

        Nothing drives it afterwards: the RTG has one state, and a light that
        never changes is exactly the reading the module is worth.
        """
        kit = _artkit()
        if kit is None:
            return
        art = self.add_child(kit.build_rtg())
        art.rotation = ART_QUARTER_TURN
        for material in _accent_materials(art):
            material.emissive_strength = accent_strength() * RTG_GLOW

    def output_per_s(self) -> float:
        return balance.RTG_OUTPUT_PER_S


# ============================================================================
# The system
# ============================================================================


class PowerSystem(Node):
    """The run's capacitor, fuel tank, life support and power posture.

    Registered as the ``Services.POWER`` singleton. Every consumer spends
    through :meth:`request` (an atomic bite) or :meth:`drain` (a per-second
    draw), and a denial is an event in its own right: ``ENERGY_DENIED`` is what
    tells the HUD to flash the arc and the audio director to click.

    Per frame it burns the generator's fuel, files the generator's noise with
    the signature meter, charges the capacitor from every live source and from
    the emergency bus, and bleeds oxygen at the base rate plus a per-breach
    penalty.
    """

    capacitor_changed = Signal(float, float)
    energy_denied = Signal(str)
    o2_changed = Signal(float, float)
    fuel_changed = Signal(float, float)
    silent_running_changed = Signal(bool)

    def __init__(self, *, capacitor_max: float = balance.CAPACITOR_MAX, **kwargs):
        super().__init__(**kwargs)
        self.capacitor_max = float(capacitor_max)
        self.capacitor = self.capacitor_max
        self.o2_max = balance.O2_MAX
        self.o2 = balance.O2_MAX
        self.fuel_max = balance.FUEL_MAX
        self.fuel = balance.FUEL_MAX
        self.open_breaches = 0
        self.silent_running = False
        #: Whether the posture answers its key at all. The onboarding schedule
        #: holds it back until run three; flow sets this at run start.
        self.silent_running_enabled = True
        self.biome = balance.BIOMES["debris_field"]
        self._sources: list[PowerSource] = []
        self._since_source_sweep = REWIRE_INTERVAL_S
        self._wiring = SignalWiring(self)
        self._elapsed = 0.0
        #: ``consumer -> elapsed`` of its last ``ENERGY_DENIED`` emission.
        self._denied_at: dict[str, float] = {}
        #: Seconds left of a forced blackout. See :meth:`begin_blackout`.
        self.blackout_remaining = 0.0

    # --- Lifecycle ---------------------------------------------------------

    def on_ready(self):
        self._wiring.want(SignalNames.BREACH_OPENED, self._on_breach_count)
        self._wiring.want(SignalNames.BREACH_PATCHED, self._on_breach_count)
        self._wiring.want(SignalNames.SECTOR_ENTERED, self._on_sector_entered)
        self._wiring.sweep()
        self.refresh_sources()

    def on_update(self, dt: float):
        self._elapsed += dt
        self._wiring.poll(dt)
        self._since_source_sweep += dt
        if self._since_source_sweep >= REWIRE_INTERVAL_S:
            self._since_source_sweep = 0.0
            self.refresh_sources()
        if Input.is_action_just_pressed("silent_running"):
            if self.silent_running_enabled:
                self.set_silent_running(not self.silent_running)
            else:
                self._answer_locked_silent_running()
        self._tick_blackout(dt)
        self._tick_fuel(dt)
        self._tick_signature(dt)
        self._tick_generation(dt)
        self._tick_life_support(dt)

    # --- Blackout ----------------------------------------------------------

    def begin_blackout(self, seconds: float) -> None:
        """Empty the capacitor and hold it empty for *seconds*.

        The Shrike's lantern burn is the one caller. It used to do this by
        spending the whole charge through :meth:`request` every frame, which
        emptied the tank but let every source refill it in between, so the arc
        strobed instead of going out and the blackout was invisible even though
        the number was right. One state, one reading: while this is running the
        bar is at zero, :attr:`blacked_out` is true, and the HUD can grey the
        arc rather than draw a live gauge that happens to be empty.
        """
        seconds = float(seconds)
        if seconds <= 0.0:
            return
        self.blackout_remaining = max(self.blackout_remaining, seconds)
        self._set_capacitor(0.0)

    @property
    def blacked_out(self) -> bool:
        """True while a blackout is holding the capacitor at zero."""
        return self.blackout_remaining > 0.0

    def _tick_blackout(self, dt: float) -> None:
        if self.blackout_remaining <= 0.0:
            return
        self.blackout_remaining = max(0.0, self.blackout_remaining - dt)
        self._set_capacitor(0.0)

    # --- Sources -----------------------------------------------------------

    def refresh_sources(self) -> None:
        """Rediscover the socketed power modules currently in the tree.

        Called on an interval and directly by ``modules.py`` after a socket
        change, so a freshly installed RTG pays from the next frame.
        """
        tree = self.tree
        if tree is None:
            self._sources = []
            return
        found: list[PowerSource] = []
        for node in iter_tree_nodes(tree):
            if isinstance(node, PowerSource):
                node.set_biome(self.biome)
                found.append(node)
        self._sources = found

    def sources(self) -> list[PowerSource]:
        """The power modules the system is currently drawing from."""
        return list(self._sources)

    def _live_sources(self) -> list[PowerSource]:
        if not self.silent_running:
            return self._sources
        return [source for source in self._sources if source.silent_safe]

    def generation_per_s(self) -> float:
        """Total energy per second from wings, generator and RTG right now."""
        return sum(source.output_per_s() for source in self._live_sources())

    def emergency_trickle_per_s(self) -> float:
        """The hull's own charge rate, owed to no module and stopped by nothing.

        Silent running does not touch it and neither does an empty fuel tank: a
        ship with folded wings, a dry generator and a flat capacitor still
        climbs back to firing on this alone. It sits far below every weapon's
        draw, so it buys the way out of a dead stop and never the fight.
        """
        return POWER_EMERGENCY_TRICKLE_PER_S

    def charge_rate_per_s(self) -> float:
        """Everything reaching the capacitor per second, modules and bus alike."""
        return self.generation_per_s() + self.emergency_trickle_per_s()

    # --- Per-frame ticks ---------------------------------------------------

    def _tick_fuel(self, dt: float) -> None:
        burners = [source for source in self._live_sources() if source.fuel_per_s() > 0.0]
        if not burners:
            return
        needed = sum(source.fuel_per_s() for source in burners) * dt
        if needed <= self.fuel:
            self._set_fuel(self.fuel - needed)
            return
        self._set_fuel(0.0)
        for source in burners:
            source.shut_down()

    def _tick_signature(self, dt: float) -> None:
        meter = self._meter()
        if meter is None:
            return
        for source in self._live_sources():
            gain = source.signature_per_s()
            if gain > 0.0:
                meter.add(gain * dt, source.signature_reason)

    def _tick_generation(self, dt: float) -> None:
        if self.blacked_out:
            return  # the bus is down; the sources are still turning, into nothing
        self.add_energy(self.charge_rate_per_s() * dt)

    def _tick_life_support(self, dt: float) -> None:
        drain = balance.O2_DRAIN_PER_S + self.open_breaches * balance.O2_DRAIN_PER_BREACH_PER_S
        self._set_o2(self.o2 - drain * dt)

    # --- Energy ------------------------------------------------------------

    def request(self, amount: float, consumer: str) -> bool:
        """Spend *amount*. False and ``ENERGY_DENIED`` if the tank is short.

        Weapons spend against a floor instead: a shot is refused only below
        :data:`WEAPONS_ENERGY_FLOOR`, and above it goes off for whatever the
        capacitor still holds. Losing a fight because the tank was a fraction
        under the exact cost reads as a broken gun rather than as a resource
        decision, and the denial that matters is the one the player can see
        coming. Every other consumer spends atomically or not at all.

        A denial is an event in its own right: ``ENERGY_DENIED`` names the
        consumer so the HUD and the audio director can say "no power" rather
        than the "no ammo" an empty magazine says. The cue is rate-limited per
        consumer to one per :data:`ENERGY_DENIED_CUE_INTERVAL_S`, because a
        held beam over an empty tank asks every frame; the request itself is
        still refused every time.
        """
        amount = float(amount)
        if amount <= 0.0:
            return True
        floor = WEAPONS_ENERGY_FLOOR if is_weapon_consumer(consumer) else amount
        if self.capacitor < floor:
            self._deny(consumer)
            return False
        self._set_capacitor(self.capacitor - amount)
        return True

    def _deny(self, consumer: str) -> None:
        """Emit the denial cue, at most once per interval per consumer."""
        last = self._denied_at.get(consumer)
        if last is not None and self._elapsed - last < ENERGY_DENIED_CUE_INTERVAL_S:
            return
        self._denied_at[consumer] = self._elapsed
        self.energy_denied(consumer)

    def drain(self, per_second: float, dt: float, consumer: str) -> bool:
        """Spend a per-second draw for one frame. Same contract as :meth:`request`.

        A beam therefore gets the frame that takes the capacitor under the floor
        and stutters out on the next one, which is the same "one last shot" rule
        the pulse weapons get.
        """
        return self.request(per_second * dt, consumer)

    def add_energy(self, amount: float) -> None:
        """Charge the capacitor, clamped to its maximum."""
        self._set_capacitor(self.capacitor + float(amount))

    def set_capacitor_max(self, maximum: float) -> None:
        """Raise or lower the capacitor ceiling (hull choice, doctrine, modules)."""
        self.capacitor_max = float(maximum)
        self._set_capacitor(self.capacitor)

    def _set_capacitor(self, value: float) -> None:
        value = min(self.capacitor_max, max(0.0, value))
        if value == self.capacitor:
            return
        self.capacitor = value
        self.capacitor_changed(value, self.capacitor_max)

    # --- Fuel and oxygen ---------------------------------------------------

    def add_fuel(self, amount: float) -> None:
        """Take on fuel from a comet, a wreck tank or a depot purchase."""
        self._set_fuel(self.fuel + float(amount))

    def spend_fuel(self, amount: float) -> bool:
        """Spend *amount* of fuel atomically. False if the tank cannot cover it."""
        amount = float(amount)
        if amount <= 0.0:
            return True
        if self.fuel < amount:
            return False
        self._set_fuel(self.fuel - amount)
        return True

    def add_o2(self, amount: float) -> None:
        """Refill life support from an ice chunk or a depot canister."""
        self._set_o2(self.o2 + float(amount))

    def restore_levels(self, *, capacitor: float, fuel: float, o2: float) -> None:
        """Adopt saved meter levels wholesale, clamped to the current maxima.

        The resume path after a suspended run: a rebuilt system starts full,
        and the levels the player actually had are applied in one step so every
        change signal fires exactly once per meter.
        """
        self._set_capacitor(float(capacitor))
        self._set_fuel(float(fuel))
        self._set_o2(float(o2))

    def _set_fuel(self, value: float) -> None:
        value = min(self.fuel_max, max(0.0, value))
        if value == self.fuel:
            return
        self.fuel = value
        self.fuel_changed(value, self.fuel_max)

    def _set_o2(self, value: float) -> None:
        value = min(self.o2_max, max(0.0, value))
        if value == self.o2:
            return
        self.o2 = value
        self.o2_changed(value, self.o2_max)

    # --- Posture -----------------------------------------------------------

    def set_silent_running(self, active: bool) -> None:
        """Go dark, or come back up.

        Going dark shuts every source that is not silent-safe, so the generator
        stops humming and stops burning fuel; the signature meter picks the
        posture up from the same frame and begins decaying toward the act floor.
        """
        active = bool(active)
        if active == self.silent_running:
            return
        self.silent_running = active
        if active:
            for source in self._sources:
                source.shut_down()
        self.silent_running_changed(active)
        self._meter()

    def _answer_locked_silent_running(self) -> None:
        """Answer C on a run flown before the posture unlocks.

        The key does nothing, so the answer goes on the toast line; going dark
        anyway with the HUD light hidden would be a mechanic the player cannot
        see acting on a run that has not taught it.
        """
        tree = self.tree
        hud = tree.singletons.get(Services.HUD) if tree is not None else None
        if hud is not None:
            hud.show_toast(SILENT_RUNNING_LOCKED_TOAST)

    def set_biome(self, biome_id: str) -> None:
        """Adopt the sector's biome, which sets how much light the wings get."""
        self.biome = balance.BIOMES[biome_id]
        for source in self._sources:
            source.set_biome(self.biome)

    # --- Signal handlers ---------------------------------------------------

    def _on_breach_count(self, open_breaches: int) -> None:
        self.open_breaches = int(open_breaches)

    def _on_sector_entered(self, sector_index: int, biome_id: str) -> None:
        self.set_biome(biome_id)

    def _meter(self):
        """The signature meter, told the current power posture as it is fetched.

        Syncing here rather than only on the toggle means a meter registered
        after the player went dark still learns the posture on the next frame.
        """
        tree = self.tree
        if tree is None:
            return None
        meter = tree.singletons.get(Services.SIGNATURE)
        if meter is not None:
            meter.set_silent_running(self.silent_running)
        return meter