shrike/ship.py

Part of SHRIKE.

"""The player ship: drift flight, the shield arc, hull breaches and the warp spool.

The ship is the hub every other gameplay module talks to. It owns four things
that nothing else may own:

* **Flight.** Dampened drift on the flight plane, with the nose aimed
  independently of the thrust vector, so strafing runs and reverse-firing
  retreats are the native vocabulary rather than an advanced trick.
* **Hull integrity.** Damage, the breach ladder (a breach opens each time the
  hull crosses another quarter of its maximum) and the rooted patch channel.
  Breaches are the O2 leak: life support consumes ``BREACH_OPENED`` and
  ``BREACH_PATCHED`` and bleeds accordingly.
* **The shield arc.** A steerable 120-degree facing that blocks fully inside
  its cone and is bare hull everywhere else. Q and E rotate it continuously on
  keyboard; on a pad it mirrors the aim vector and the bumpers nudge it away in
  60-degree steps, so aiming twice at once survives on both devices.
* **Energy arbitration.** Every consumer that spends from the shared capacitor
  goes through :meth:`PlayerShip.request_energy` / :meth:`PlayerShip.drain_energy`,
  which forward to the run's :class:`~shrike.power.PowerSystem` when one is
  registered and to a local :class:`Capacitor` when the ship is flying on its
  own. Both arbitrate the same way: weapons fire down to
  ``balance.WEAPONS_ENERGY_FLOOR`` and every other consumer spends atomically
  or is told no, rather than half-firing.

* **The warp spool's outcome contract.** A channel that starts ends on exactly
  one of :data:`SPOOL_JUMP`, :data:`SPOOL_ABORTED` or :data:`SPOOL_FAILED`, and
  every one of them is announced. It never resets in silence, it cannot be
  pushed out of reach forever by fire (:data:`WARP_SPOOL_MAX_INTERRUPTIONS`),
  and it refuses to declare a jump the tank cannot pay for.

Everything the HUD and the audio director need is a signal: hull, breaches,
afterburner state, shield absorption and breakage, and the warp spool ladder.
"""

import importlib.util
import math

from simvx.core import (
    Area3D,
    Input,
    Material,
    Mesh,
    MeshInstance3D,
    Node,
    Node3D,
    Property,
    Quat,
    Signal,
    SphereShape3D,
    Vec2,
    Vec3,
)

from . import balance
from .runtime import (
    PLANE_Y,
    CameraRig,
    Groups,
    Layers,
    Services,
    gamepad_aim_input,
    heading_to_direction,
    move_input,
)

# ============================================================================
# Numbers the design fixes by feel rather than by table, so balance.py has no
# name for them yet. They are derived from the constants that balance.py does
# carry wherever a derivation exists, and are flagged for the balance pass.
# ============================================================================

#: Hull length in world units, the unit the drift budget is quoted in.
SHIP_LENGTH_UNITS = balance.SHIP_LENGTH_UNITS
#: Cruise top speed, world units per second. Chosen so that releasing all keys
#: coasts ``balance.RELEASE_DRIFT_SHIP_LENGTHS`` hull lengths under
#: ``balance.INERTIAL_DAMPENING``: the coast is ``v / (60 * (1 - d))`` units.
#: The derivation lives in balance.py so the enemy speed table shares it.
CRUISE_SPEED = balance.PLAYER_CRUISE_SPEED
#: Thrust acceleration, world units per second squared: the value at which the
#: dampened velocity settles on ``CRUISE_SPEED`` at the 60 Hz fixed step.
THRUST_ACCELERATION = CRUISE_SPEED * 60.0 * (1.0 - balance.INERTIAL_DAMPENING) / balance.INERTIAL_DAMPENING
#: Afterburner multiplies both the acceleration and the speed ceiling.
AFTERBURNER_SPEED_MULT = balance.PLAYER_AFTERBURNER_SPEED_MULT

#: Keyboard shield rotation rate, radians per second (about 170 degrees/s).
SHIELD_ROTATE_RADIANS_PER_S = 3.0
#: Shield recovery, absorption points per second while it is drawing energy.
SHIELD_REGEN_POINTS_PER_S = 12.0

# -- Drawing the shield ------------------------------------------------------
#
# The arc was a number and a coverage test with nothing on screen, so Q and E
# were two keys that did nothing observable and a hit taken on the bare quarter
# of the hull was indistinguishable from one the shield had failed to stop. It
# is drawn now, at all times, and it turns where the pilot turns it.

#: The band's radius from the hull centre, and the bore of the band itself.
#: Outside the hull, inside the HUD's own foveal arcs, so the three rings
#: (capacitor, hull, shield) stay told apart.
SHIELD_BAND_RADIUS = 2.35
SHIELD_BAND_BORE = 0.11
#: Points along the band. A 120-degree arc needs enough of them to read as a
#: curve rather than as a bent stick.
SHIELD_BAND_POINTS = 17
#: The arc's own colour: cold blue, which the readability contract leaves free
#: (player fire is warm gold, enemy fire magenta, telegraphs white).
SHIELD_BAND_COLOUR = (0.48, 0.82, 1.0)
#: What a broken arc is drawn as while its lockout runs: still there, so the
#: pilot sees which quarter is not being covered, and unmistakably dead.
SHIELD_BROKEN_COLOUR = (0.95, 0.36, 0.30)
#: Opacity of the band at full charge and at none, plus what a fresh absorb
#: adds on top and how fast that flash decays.
SHIELD_ALPHA_FULL = 0.34
SHIELD_ALPHA_EMPTY = 0.10
SHIELD_FLASH_ALPHA = 0.85
SHIELD_FLASH_DECAY_PER_S = 3.2
#: Emissive strength at rest and at the peak of an absorb flash. Both stay
#: under the bloom threshold at rest: the arc marks a facing, it does not light
#: the sector.
SHIELD_GLOW_REST = 0.8
SHIELD_GLOW_FLASH = 3.4

#: Right-stick deflection past which the pad owns the aim.
PAD_AIM_DEADZONE = 0.25
#: Mouse travel in pixels that hands the aim back to the mouse.
MOUSE_TAKEOVER_PIXELS = 2.0
#: How far ahead of the ship the pad's aim point sits, world units.
PAD_AIM_REACH_UNITS = 22.0
#: Reach of the fallback aim mapping used before a camera exists.
FALLBACK_AIM_REACH_UNITS = 30.0

#: Where the exhaust nacelles sit in the ship's local frame. The nose runs down
#: local -Z, so the tail is +Z and the pair straddle the keel. These are the
#: mouths of the art kit's own engine bells: the Vagrant is 2.6 units long with
#: a 1.7 beam, and its two bells sit at 94 per cent of the half length behind
#: the waist, a third of the beam apart. A plume that starts anywhere else
#: reads as fire coming out of the hull rather than out of the engines.
ENGINE_NACELLE_OFFSETS = ((-0.53, 0.0, 1.22), (0.53, 0.0, 1.22))
#: Ribbon throttle reached on full thrust with the burner cold. The rest of the
#: range belongs to the afterburner, so lighting it is a visible change in the
#: plume and not just a number in the HUD.
RIBBON_CRUISE_THROTTLE = 0.65
#: How fast the plume spools up and down, in throttle fraction per second. A
#: key is either down or it is not, so without a ramp the jet appears and
#: vanishes between two frames, which reads as a flicker rather than a drive.
#: It relights faster than it dies, the way a real throttle answers.
RIBBON_SPOOL_UP_PER_S = 8.0
RIBBON_SPOOL_DOWN_PER_S = 3.4
#: Throttle a lit burner holds on its own, with no thrust key down. The burner
#: drains the capacitor whether or not the ship is being pushed, so the plume
#: has to say it is lit rather than only saying the ship is accelerating.
RIBBON_BURNER_FLOOR = 0.45

#: Hold time on the interact action that starts a breach patch, per section 10.
INTERACT_PATCH_HOLD_S = 1.0

# ---------------------------------------------------------------------------
# The warp spool's outcome contract
#
# A started channel ends on exactly one of these, and every ending is
# announced. A ring that closes and reopens under fire forever, or one that is
# reset without a word when the pilot dies, is a five-second animation that
# taught the player nothing about what their drive just did.
# ---------------------------------------------------------------------------

#: The jump happened: ``warp_completed`` fired.
SPOOL_JUMP = "jump"
#: The channel was stood down deliberately (the abort key, a tear-in, death).
SPOOL_ABORTED = "aborted"
#: The channel could not finish, and :attr:`PlayerShip.spool_failure` says why.
SPOOL_FAILED = "failed"

#: Interruptions one channel survives. Each adds
#: ``balance.WARP_SPOOL_PENALTY_PER_HIT_S``, so fleeing under fire degrades as
#: the design asks; past this it breaks, because a channel that can be pushed
#: out of reach indefinitely is not a decision the pilot can read.
WARP_SPOOL_MAX_INTERRUPTIONS = balance.WARP_SPOOL_MAX_INTERRUPTIONS
#: The two reasons a channel fails, in the words the HUD puts on screen.
SPOOL_FAILURE_UNDER_FIRE = "SPOOL BROKEN: taking fire"
SPOOL_FAILURE_NO_FUEL = "NO FUEL TO COMPLETE JUMP"

#: Radius of the ship's hitbox, world units.
HITBOX_RADIUS = 1.0

# ---------------------------------------------------------------------------
# Hulls
#
# The design says hardpoints grow "two to four by hull", and for a long time
# nothing anywhere read that sentence: ``WeaponRack.add_hardpoint`` had no
# callers at all, every rack was built at ``balance.HARDPOINTS_STARTER``, and
# the Barge, the Dart and the Hive were three unlockable trophies that flew the
# starter's two guns and wore the starter's silhouette. The per-hull count now
# lives in ``balance.HULL_HARDPOINTS``, beside ``balance.HULL_SOCKETS``.
# ---------------------------------------------------------------------------

#: The hull a run flies unless the run configuration names another.
DEFAULT_HULL = "vagrant"


def hardpoints_for_hull(hull_id: str) -> int:
    """How many gun mounts *hull_id* flies with, raising on an unknown hull."""
    try:
        return balance.HULL_HARDPOINTS[hull_id]
    except KeyError:
        raise ValueError(
            f"unknown hull {hull_id!r}; expected one of {', '.join(sorted(balance.HULL_HARDPOINTS))}"
        ) from None


#: Validation ceiling on the hull Property, not a balance number: hulls and
#: plating raise the working maximum, and this only stops a nonsense write.
HULL_PROPERTY_CEILING = 10_000.0


def wrap_angle(radians: float) -> float:
    """Fold an angle into ``[-pi, pi]``."""
    return math.remainder(float(radians), math.tau)


def bearing_of(direction: Vec3) -> float | None:
    """Heading of a world direction on the flight plane, or None if it names none."""
    dx, dz = float(direction[0]), float(direction[2])
    if math.hypot(dx, dz) < 1e-6:
        return None
    return math.atan2(-dz, dx)


class Capacitor(Node):
    """The shared energy budget, and the arbitration rule for spending it.

    Every consumer, weapons, shield recovery, the afterburner, the tractor
    scoop, auto-turrets and the Refinery, asks this object for energy instead
    of subtracting from a number of its own. The rule is all-or-nothing: a
    request larger than the stored charge spends nothing and reports the denial
    on :attr:`energy_denied`, so a beam that cannot afford a full tick stutters
    audibly rather than draining the capacitor other consumers were counting on.

    This is the standalone stand-in for the run's ``PowerSystem``, which owns
    generation as well as spending, and it keeps the two spending rules the
    real system arbitrates by. Weapons spend against
    ``balance.WEAPONS_ENERGY_FLOOR`` rather than the exact cost: a shot is
    refused only below the floor, and above it goes off for whatever charge is
    left, so a fight never ends because the tank was a fraction short. And the
    hull's emergency bus (``balance.POWER_EMERGENCY_TRICKLE_PER_S``) keeps a
    trickle flowing, so a ship flying without a power system is dim rather
    than bricked. Socketed generation still needs the real system: the ship
    builds this only when no ``Services.POWER`` singleton is registered.
    """

    capacitor_changed = Signal(float, float)
    energy_denied = Signal(str)

    def __init__(self, capacity: float = balance.CAPACITOR_MAX, **kwargs):
        super().__init__(**kwargs)
        self.capacitor_max = float(capacity)
        self.capacitor = float(capacity)
        #: Name of the consumer refused most recently, for the HUD's denial flash.
        self.last_denied: str | None = None

    def on_update(self, dt: float):
        self.add_energy(balance.POWER_EMERGENCY_TRICKLE_PER_S * dt)

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

        Same contract as ``PowerSystem.request``: weapons are refused only
        below ``balance.WEAPONS_ENERGY_FLOOR`` and spend whatever is left;
        every other consumer spends atomically or not at all.
        """
        amount = float(amount)
        if amount <= 0.0:
            return True
        floor = balance.WEAPONS_ENERGY_FLOOR if consumer in balance.WEAPONS else amount - 1e-9
        if self.capacitor < floor:
            self.last_denied = consumer
            self.energy_denied(consumer)
            return False
        self.capacitor = max(0.0, self.capacitor - amount)
        self.capacitor_changed(self.capacitor, self.capacitor_max)
        return True

    def drain(self, per_second: float, dt: float, consumer: str) -> bool:
        """Spend a per-second rate over *dt*, on the same all-or-nothing rule."""
        return self.request(float(per_second) * float(dt), consumer)

    def add_energy(self, amount: float) -> None:
        """Add charge, clamped to the capacitor's ceiling."""
        amount = float(amount)
        if amount <= 0.0:
            return
        self.capacitor = min(self.capacitor_max, self.capacitor + amount)
        self.capacitor_changed(self.capacitor, self.capacitor_max)


class ShieldArc(Node3D):
    """The steerable 120-degree shield facing, a child of :class:`PlayerShip`.

    The arc blocks fully inside its cone and not at all outside it, so defence
    is a second aiming problem laid over the first. It absorbs
    ``balance.SHIELD_ABSORB_MAX`` points before it breaks, stays down for
    ``balance.SHIELD_BREAK_LOCKOUT_S``, and then recovers while it can pay
    ``balance.ENERGY_SHIELD_REGEN_PER_S`` out of the capacitor. Recovery stalls,
    it does not fail, when the capacitor is empty.

    :attr:`arc_centre` is measured from the ship's nose, so it costs nothing to
    hold the shield off-axis while the nose tracks a target: a pad player nudges
    the offset in 60-degree steps, a keyboard player sweeps it with Q and E, and
    both leave the aim untouched.
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        #: Arc centre in radians, measured anticlockwise from the ship's nose.
        self.arc_centre: float = 0.0
        self.charge: float = balance.SHIELD_ABSORB_MAX
        self.broken: bool = False
        self.lockout_remaining: float = 0.0
        self.half_width: float = math.radians(balance.SHIELD_ARC_DEGREES) * 0.5
        #: The drawn band, mounted on ready. The node's own local -Z is the
        #: arc's centre, so its ``forward`` reads back as the covered bearing.
        self.band: MeshInstance3D | None = None
        self._ship: PlayerShip | None = None
        self._flash = 0.0

    @property
    def charge_max(self) -> float:
        return balance.SHIELD_ABSORB_MAX

    @property
    def charge_fraction(self) -> float:
        """How much of the arc's absorption is left, 0 to 1."""
        return self.charge / self.charge_max if self.charge_max > 0.0 else 0.0

    # -- the drawn arc ----------------------------------------------------

    def on_ready(self):
        self.band = self.add_child(
            MeshInstance3D(
                name="Band",
                mesh=Mesh.extrude_path(self._band_centreline(), sides=6, radius=SHIELD_BAND_BORE),
                material=Material(
                    colour=(*SHIELD_BAND_COLOUR, SHIELD_ALPHA_FULL),
                    emissive_colour=SHIELD_BAND_COLOUR,
                    emissive_strength=SHIELD_GLOW_REST,
                ),
            )
        )
        self._refresh_band()

    def _band_centreline(self) -> list[tuple[float, float, float]]:
        """The arc's points, swept about the node's own local -Z.

        Local -Z is what :meth:`Node3D.face_along` aims and what the parent
        hull's nose already runs down, so laying the band around it and turning
        the node by :attr:`arc_centre` puts the drawn arc exactly where
        :meth:`covers` says the cover is.
        """
        points = []
        for index in range(SHIELD_BAND_POINTS):
            offset = -self.half_width + 2.0 * self.half_width * index / (SHIELD_BAND_POINTS - 1)
            points.append(
                (
                    -SHIELD_BAND_RADIUS * math.sin(offset),
                    0.0,
                    -SHIELD_BAND_RADIUS * math.cos(offset),
                )
            )
        return points

    def _refresh_band(self) -> None:
        """Turn the band onto the covered bearing and colour it for its state."""
        band = self.band
        if band is None:
            return
        self.rotation = Quat.from_axis_angle(Vec3(0.0, 1.0, 0.0), self.arc_centre)
        colour = SHIELD_BROKEN_COLOUR if self.broken else SHIELD_BAND_COLOUR
        rest = SHIELD_ALPHA_EMPTY + (SHIELD_ALPHA_FULL - SHIELD_ALPHA_EMPTY) * self.charge_fraction
        alpha = rest + (SHIELD_FLASH_ALPHA - rest) * self._flash
        material = band.material
        material.colour = (*colour, alpha)
        material.emissive_colour = colour
        material.emissive_strength = SHIELD_GLOW_REST + (SHIELD_GLOW_FLASH - SHIELD_GLOW_REST) * self._flash

    def absolute_centre(self) -> float:
        """The arc's centre as a world heading, folding in the ship's nose."""
        nose = self._ship.heading if self._ship is not None else 0.0
        return wrap_angle(nose + self.arc_centre)

    def covers(self, incoming_direction: Vec3) -> bool:
        """True if the arc faces *incoming_direction*.

        *incoming_direction* is the world-space direction from the ship toward
        the source of the damage, the bearing under attack. A broken arc covers
        nothing, and a zero-length direction names no bearing, so it is not
        covered either.
        """
        if self.broken:
            return False
        bearing = bearing_of(Vec3(incoming_direction))
        if bearing is None:
            return False
        return abs(wrap_angle(bearing - self.absolute_centre())) <= self.half_width + 1e-9

    def absorb(self, amount: float) -> float:
        """Soak up to the remaining charge and return the damage that got through.

        Coverage is the caller's question: :meth:`covers` answers it, and
        :meth:`PlayerShip.apply_damage` asks it. A hit that empties the charge
        breaks the arc and starts the lockout.
        """
        amount = float(amount)
        if self.broken or amount <= 0.0:
            return max(0.0, amount)
        soaked = min(amount, self.charge)
        self.charge -= soaked
        if soaked > 0.0:
            self._flash = 1.0
        if self.charge <= 1e-9:
            self.charge = 0.0
            self.broken = True
            self.lockout_remaining = balance.SHIELD_BREAK_LOCKOUT_S
        return amount - soaked

    def nudge(self, delta: float) -> None:
        """Rotate the arc by *delta* radians relative to the nose."""
        self.arc_centre = wrap_angle(self.arc_centre + float(delta))

    def recentre(self) -> None:
        """Snap the arc back onto the nose, which on a pad is the aim vector."""
        self.arc_centre = 0.0

    def on_update(self, dt: float):
        self._flash = max(0.0, self._flash - SHIELD_FLASH_DECAY_PER_S * dt)
        self._refresh_band()
        if self.lockout_remaining > 0.0:
            self.lockout_remaining = max(0.0, self.lockout_remaining - dt)
            if self.lockout_remaining > 0.0:
                return
            self.broken = False
        if self.charge >= self.charge_max or self._ship is None:
            return
        if not self._ship.drain_energy(balance.ENERGY_SHIELD_REGEN_PER_S, dt, "shield_regen"):
            return
        self.charge = min(self.charge_max, self.charge + SHIELD_REGEN_POINTS_PER_S * dt)


class PlayerShip(Node3D):
    """The player's hull: drift flight, aim, breaches, shields and the warp spool.

    Flight is the design's 30-second loop. Thrust accelerates the plane velocity
    and ``balance.INERTIAL_DAMPENING`` bleeds it, so releasing everything coasts
    about ``balance.RELEASE_DRIFT_SHIP_LENGTHS`` hull lengths; the nose points at
    the aim point regardless of where the thrust is pushing. The afterburner
    lifts both the acceleration and the ceiling over
    ``balance.AFTERBURNER_IGNITION_RAMP_S`` while it can pay
    ``balance.ENERGY_AFTERBURNER_PER_S``, and cuts out the moment it cannot.

    Damage lands through :meth:`apply_damage`, which is the only path into the
    hull. It resolves the shield arc first (combat's router must not resolve it
    a second time), lengthens a running warp spool, and opens a breach each time
    the hull crosses another ``balance.BREACH_HULL_FRACTION_STEP`` of its
    maximum. Breaches never close on their own: :meth:`patch_breach` roots the
    ship for ``balance.BREACH_PATCH_CHANNEL_S`` and costs
    ``balance.BREACH_PATCH_SCRAP``.
    """

    hull = Property(
        balance.HULL_MAX_STARTER,
        range=(0.0, HULL_PROPERTY_CEILING),
        on_change="_on_hull_changed",
        hint="Current hull integrity",
    )
    hull_max = Property(
        balance.HULL_MAX_STARTER,
        range=(1.0, HULL_PROPERTY_CEILING),
        hint="Hull integrity ceiling; the hull choice may raise it",
    )

    # Signals, named exactly as runtime.SignalNames registers them.
    hull_changed = Signal(float, float)
    breach_opened = Signal(int)
    breach_patched = Signal(int)
    ship_destroyed = Signal()
    afterburner_changed = Signal(bool)
    shield_absorbed = Signal(float)
    shield_broken = Signal()
    warp_spool_started = Signal(bool)
    warp_spool_interrupted = Signal(float)
    warp_spool_cancelled = Signal()
    warp_completed = Signal(bool)

    # Class-level defaults: the hull Property's change hook can fire before
    # __init__ has finished assigning instance state.
    open_breaches: int = 0
    _breaches_seen: int = 0
    _destroyed: bool = False

    def __init__(self, *, hull_id: str = DEFAULT_HULL, **kwargs):
        super().__init__(**kwargs)
        #: Which of the four flyable hulls this is. It decides the silhouette
        #: the art kit builds and how many gun mounts the rack should open.
        self.hull_id = str(hull_id)
        hardpoints_for_hull(self.hull_id)  # Fail on the constructor, not on the art.
        #: Plane velocity, ``Vec2(x, z)`` in world units per second.
        self.velocity = Vec2(0.0, 0.0)
        #: Nose heading in radians, anticlockwise from +X seen from above.
        self.heading: float = math.pi * 0.5
        self.open_breaches = 0
        self._breaches_seen = 0
        self._destroyed = False

        self.shield: ShieldArc | None = None
        self.assists = balance.AssistSettings()
        #: Set by the hunter for the arrival lockout: the drive will not spool.
        self.warp_scrambled = False

        self._capacitor: Capacitor | None = None
        self._rig: CameraRig | None = None
        self._ribbons: list[Node3D] = []
        self._ribbon_throttle = 0.0
        #: The drive state the plumes were last given, so a coasting ship does
        #: not rewrite two materials and four emitter properties per ribbon per
        #: frame to arrive at the plume already on the tail.
        self._ribbon_drive: tuple | None = None
        self._thrust_input = Vec2(0.0, 0.0)
        self._afterburner_ramp = 0.0
        self._afterburner_active = False
        self._patch_remaining = 0.0
        self._spooling = False
        self._spool_remaining = 0.0
        self._spool_emergency = False
        self._spool_outcome = ""
        self._spool_failure = ""
        self._spool_interruptions = 0
        self._interact_hold = 0.0
        self._interact_consumed = False
        self._pad_aim = False
        self._aim_direction = heading_to_direction(self.heading)
        self._last_mouse = Vec2(Input.mouse_position)

    # ------------------------------------------------------------------ setup

    def on_enter_tree(self):
        super().on_enter_tree()
        self.add_to_group(Groups.SHIP)

    def on_ready(self):
        self.position = Vec3(float(self.position.x), PLANE_Y, float(self.position.z))
        self.add_child(self._build_hull_visual())
        self.add_child(
            Area3D(
                name="Hitbox",
                shape=SphereShape3D(radius=HITBOX_RADIUS),
                collision_layer=Layers.SHIP,
                collision_mask=Layers.MASK_SHIP,
            )
        )
        self.shield = self.add_child(ShieldArc(name="ShieldArc"))
        self.shield._ship = self
        self._mount_ribbons()
        self._face_nose()

    def _mount_ribbons(self) -> None:
        """Hang an exhaust ribbon on each nacelle, where the vfx layer exists.

        The plume is the ship's own answer to a screen with nothing else on it:
        it says the drive is lit, which way it is pushing and, through the
        ribbon's notoriety hook, how loud the pilot's reputation has become.
        """
        if importlib.util.find_spec("shrike.vfx") is None:
            return
        from .vfx import EngineRibbon

        for index, offset in enumerate(ENGINE_NACELLE_OFFSETS):
            self._ribbons.append(self.add_child(EngineRibbon(name=f"EngineRibbon{index}", position=Vec3(*offset))))

    def _build_hull_visual(self) -> Node3D:
        """The hull geometry, mounted so its nose runs down the node's local -Z.

        That is the axis :meth:`Node3D.face_along` aims, so the mount is the one
        place the art's own convention is reconciled with the scene's. The art
        kit builds the ship pointing along +X, and a quarter turn about Y brings
        it onto -Z. Until ``shrike.artkit`` is importable the ship flies as a
        plain cone, which is enough to read a heading on screen.
        """
        mount = Node3D(name="Hull")
        if importlib.util.find_spec("shrike.artkit") is not None:
            from .artkit import build_player_ship

            art = build_player_ship(hull=self.hull_id, hardpoints=self.hardpoint_count)
            art.rotation = Quat.from_axis_angle(Vec3(0.0, 1.0, 0.0), math.pi * 0.5)
            mount.add_child(art)
        else:
            mount.add_child(
                MeshInstance3D(
                    name="HullMesh",
                    mesh=Mesh.cone(radius=0.6, height=SHIP_LENGTH_UNITS, segments=12),
                    material=Material(colour=(0.75, 0.78, 0.82, 1.0), metallic=0.8, roughness=0.35),
                    rotation=Quat.from_euler(math.radians(90), 0.0, 0.0),
                )
            )
        return mount

    @property
    def hardpoint_count(self) -> int:
        """Gun mounts this hull opens, the size the run's ``WeaponRack`` is built at.

        The rack is assembled by the run scene, so this is where it asks; the
        hull answers once and both the rack and the pylons on the hull mesh come
        from the same number.
        """
        return hardpoints_for_hull(self.hull_id)

    # ------------------------------------------------------------------ energy

    @property
    def power(self):
        """The energy source this ship spends from.

        The run's ``PowerSystem`` when one is registered, otherwise a local
        :class:`Capacitor` built on first use. Both expose the same
        ``request`` / ``drain`` / ``add_energy`` arbitration surface, so no
        consumer needs to know which it is talking to.
        """
        tree = self.tree
        if tree is not None:
            system = tree.singletons.get(Services.POWER)
            if system is not None:
                return system
        if self._capacitor is None:
            self._capacitor = self.add_child(Capacitor(name="Capacitor"))
        return self._capacitor

    def request_energy(self, amount: float, consumer: str) -> bool:
        """Spend *amount* from the capacitor atomically, naming the consumer."""
        return bool(self.power.request(amount, consumer))

    def drain_energy(self, per_second: float, dt: float, consumer: str) -> bool:
        """Spend a per-second rate over *dt*, atomically, naming the consumer."""
        return bool(self.power.drain(per_second, dt, consumer))

    # -------------------------------------------------------------------- aim

    def aim_point(self) -> Vec3:
        """The point on the flight plane the player is aiming at.

        The right stick owns the aim whenever it is deflected, and keeps it
        while it rests at centre so a released stick does not snap the nose onto
        a stale mouse position. Any real mouse movement takes it back.
        """
        pad = gamepad_aim_input()
        px, pz = float(pad.x), float(pad.y)
        if math.hypot(px, pz) >= PAD_AIM_DEADZONE:
            self._pad_aim = True
            length = math.hypot(px, pz)
            self._aim_direction = Vec3(px / length, 0.0, pz / length)
        else:
            mouse = Vec2(Input.mouse_position)
            moved = math.hypot(float(mouse.x) - float(self._last_mouse.x), float(mouse.y) - float(self._last_mouse.y))
            if moved > MOUSE_TAKEOVER_PIXELS:
                self._pad_aim = False
            self._last_mouse = mouse
            if not self._pad_aim:
                return self._mouse_aim_point()
        return Vec3(
            float(self.position.x) + float(self._aim_direction.x) * PAD_AIM_REACH_UNITS,
            PLANE_Y,
            float(self.position.z) + float(self._aim_direction.z) * PAD_AIM_REACH_UNITS,
        )

    def _mouse_aim_point(self) -> Vec3:
        """Cast the mouse through the camera onto the flight plane."""
        camera = self._rig.camera if self._rig is not None else None
        width, height = self.tree.screen_size if self.tree is not None else (1.0, 1.0)
        width, height = max(float(width), 1.0), max(float(height), 1.0)
        mouse = Input.mouse_position
        if camera is None:
            # No camera yet: a screen-centre-relative mapping still aims.
            dx = (float(mouse.x) - width * 0.5) / width
            dz = (float(mouse.y) - height * 0.5) / height
            return Vec3(
                float(self.position.x) + dx * FALLBACK_AIM_REACH_UNITS,
                PLANE_Y,
                float(self.position.z) + dz * FALLBACK_AIM_REACH_UNITS,
            )
        ndc_x = 2.0 * float(mouse.x) / width - 1.0
        ndc_y = 1.0 - 2.0 * float(mouse.y) / height
        half = math.tan(math.radians(float(camera.fov)) * 0.5)
        forward, right, up = camera.forward, camera.right, camera.up
        direction = Vec3(
            float(forward.x) + float(right.x) * ndc_x * half * (width / height) + float(up.x) * ndc_y * half,
            float(forward.y) + float(right.y) * ndc_x * half * (width / height) + float(up.y) * ndc_y * half,
            float(forward.z) + float(right.z) * ndc_x * half * (width / height) + float(up.z) * ndc_y * half,
        )
        origin = camera.world_position
        if float(direction.y) > -1e-6:
            # Aiming at or above the horizon: hold the last aim rather than
            # throwing the reticle to infinity behind the camera.
            return Vec3(
                float(self.position.x) + float(self._aim_direction.x) * PAD_AIM_REACH_UNITS,
                PLANE_Y,
                float(self.position.z) + float(self._aim_direction.z) * PAD_AIM_REACH_UNITS,
            )
        distance = (PLANE_Y - float(origin.y)) / float(direction.y)
        return Vec3(
            float(origin.x) + float(direction.x) * distance,
            PLANE_Y,
            float(origin.z) + float(direction.z) * distance,
        )

    def _face_nose(self) -> None:
        self.face_along(heading_to_direction(self.heading))

    # ------------------------------------------------------------------ damage

    def apply_damage(self, amount: float, direction: Vec3, kind: str = "impact") -> None:
        """Take a hit from *direction*, the bearing from the ship to its source.

        The shield arc resolves here and nowhere else. Any hit landing during a
        warp spool lengthens the channel by
        ``balance.WARP_SPOOL_PENALTY_PER_HIT_S``, blocked or not: fleeing under
        fire is meant to degrade.
        """
        amount = float(amount) * float(self.assists.damage_taken_mult)
        if self._destroyed or amount <= 0.0:
            return

        leak = amount
        if self.shield is not None and self.shield.covers(direction):
            was_broken = self.shield.broken
            leak = self.shield.absorb(amount)
            soaked = amount - leak
            if soaked > 0.0:
                self.shield_absorbed(soaked)
            if self.shield.broken and not was_broken:
                self.shield_broken()

        if self._spooling and not self.assists.spool_immunity:
            self._spool_interruptions += 1
            self._spool_remaining += balance.WARP_SPOOL_PENALTY_PER_HIT_S
            self.warp_spool_interrupted(balance.WARP_SPOOL_PENALTY_PER_HIT_S)
            if self._spool_interruptions > WARP_SPOOL_MAX_INTERRUPTIONS:
                self._end_spool(SPOOL_FAILED, SPOOL_FAILURE_UNDER_FIRE)

        if leak > 0.0:
            self.hull = max(0.0, float(self.hull) - leak)

    def _on_hull_changed(self) -> None:
        self.hull_changed(float(self.hull), float(self.hull_max))
        self._update_breaches()
        if float(self.hull) <= 0.0 and not self._destroyed:
            self._destroyed = True
            self._patch_remaining = 0.0
            self.velocity = Vec2(0.0, 0.0)
            if self._spooling:
                # Dropping the flag here and saying nothing left the ring
                # closing over a dead ship and the chart holding a route it
                # would never be told about.
                self._end_spool(SPOOL_ABORTED)
            self.ship_destroyed()

    def _update_breaches(self) -> None:
        """Open a breach for every fresh quarter of the hull that has gone."""
        maximum = max(float(self.hull_max), 1e-6)
        fraction = max(0.0, float(self.hull)) / maximum
        steps = int(1.0 / balance.BREACH_HULL_FRACTION_STEP) - 1
        crossed = min(steps, int((1.0 - fraction) / balance.BREACH_HULL_FRACTION_STEP + 1e-9))
        while crossed > self._breaches_seen:
            self._breaches_seen += 1
            self.open_breaches += 1
            self.breach_opened(self.open_breaches)

    def patch_breach(self) -> bool:
        """Start the rooted patch channel. False if there is nothing to patch.

        Costs ``balance.BREACH_PATCH_SCRAP`` where an economy is running, and
        roots the ship for ``balance.BREACH_PATCH_CHANNEL_S``: the whole point
        of the breach is that closing it costs seconds you wanted to spend
        mining.
        """
        if self._destroyed or self.open_breaches <= 0 or self._patch_remaining > 0.0:
            return False
        tree = self.tree
        economy = tree.singletons.get(Services.ECONOMY) if tree is not None else None
        if economy is not None and not economy.spend_scrap(balance.BREACH_PATCH_SCRAP):
            return False
        self._patch_remaining = balance.BREACH_PATCH_CHANNEL_S
        self.velocity = Vec2(0.0, 0.0)
        return True

    def seal_breach(self) -> bool:
        """Close one open breach at once. False when none is open.

        The rooted, scrap-priced channel is :meth:`patch_breach`; this is the
        depot's breach foam, which was paid for at the counter and works the
        moment it is aboard.
        """
        if self._destroyed or self.open_breaches <= 0:
            return False
        self.open_breaches -= 1
        self.breach_patched(self.open_breaches)
        return True

    @property
    def patching(self) -> bool:
        """True while the patch channel roots the ship."""
        return self._patch_remaining > 0.0

    def patch_fraction(self) -> float:
        """How far the patch channel has run, 0 to 1, for the HUD's radial fill."""
        if self._patch_remaining <= 0.0:
            return 0.0
        return 1.0 - self._patch_remaining / balance.BREACH_PATCH_CHANNEL_S

    # -------------------------------------------------------------------- warp

    def begin_warp_spool(self, *, emergency: bool = False) -> None:
        """Start the warp channel. A scrambled drive refuses to spool at all."""
        if self._destroyed or self._spooling or self.warp_scrambled:
            return
        self._spooling = True
        self._spool_emergency = emergency
        self._spool_remaining = balance.WARP_SPOOL_S
        self._spool_interruptions = 0
        self._spool_outcome = ""
        self._spool_failure = ""
        self.warp_spool_started(emergency)

    def cancel_warp_spool(self) -> None:
        """Abandon a running channel without completing it.

        Emits ``warp_spool_cancelled`` so the ring, the HUD label and any route
        the chart had already bought all stand down; a drive that was not
        spooling has nothing to cancel and stays silent.
        """
        if not self._spooling:
            return
        self._end_spool(SPOOL_ABORTED)

    def _end_spool(self, outcome: str, reason: str = "") -> None:
        """Close a running channel on exactly one outcome, and say which.

        This is the only way out of :attr:`spooling`. A jump fires
        ``warp_completed``; an abort and a failure both fire
        ``warp_spool_cancelled``, because to every consumer holding a route
        they are the same instruction, and a failure additionally hands the HUD
        the reason so the pilot is told what the drive could not do.
        """
        self._spooling = False
        self._spool_remaining = 0.0
        self._spool_interruptions = 0
        self._spool_outcome = outcome
        self._spool_failure = reason if outcome == SPOOL_FAILED else ""
        if outcome == SPOOL_JUMP:
            self.warp_completed(self._spool_emergency)
            return
        self.warp_spool_cancelled()
        if outcome != SPOOL_FAILED:
            return
        hud = self.tree.singletons.get(Services.HUD) if self.tree is not None else None
        if hud is not None and hasattr(hud, "fail_warp_spool"):
            hud.fail_warp_spool(reason)

    def _fuel_is_dry(self) -> bool:
        """Whether the tank ran out while the channel was running.

        The price of a given jump belongs to the chart, which refuses its own
        route when the tank cannot cover it. What the ship owns is the floor:
        an empty tank cannot buy any jump at all, so the channel fails here
        rather than announcing a jump nothing can pay for.
        """
        power = self.tree.singletons.get(Services.POWER) if self.tree is not None else None
        fuel = getattr(power, "fuel", None)
        return fuel is not None and float(fuel) <= 0.0

    @property
    def spooling(self) -> bool:
        return self._spooling

    @property
    def spool_outcome(self) -> str:
        """How the last channel ended: one of the ``SPOOL_*`` ids, or empty."""
        return self._spool_outcome

    @property
    def spool_failure(self) -> str:
        """Why the last channel failed, empty unless it did."""
        return self._spool_failure

    def spool_fraction(self) -> float:
        """How far the spool has closed, 0 to 1, for the diegetic warp ring."""
        if not self._spooling:
            return 0.0
        total = max(balance.WARP_SPOOL_S, self._spool_remaining)
        return 1.0 - self._spool_remaining / total

    @property
    def afterburner_active(self) -> bool:
        return self._afterburner_active

    @property
    def engine_ribbons(self) -> list[Node3D]:
        """The mounted exhaust plumes, empty when the vfx layer is absent."""
        return list(self._ribbons)

    # ------------------------------------------------------------------ frames

    def on_fixed_update(self, dt: float):
        if self._destroyed:
            self.velocity = Vec2(0.0, 0.0)
            self._thrust_input = Vec2(0.0, 0.0)
            return
        if self.patching:
            # Rooted: the patch is worth exactly as much as the seconds it costs.
            self.velocity = Vec2(0.0, 0.0)
            self._thrust_input = Vec2(0.0, 0.0)
            return

        thrust = move_input()
        self._thrust_input = thrust
        burn = self._tick_afterburner(dt)
        acceleration = THRUST_ACCELERATION * burn
        ceiling = CRUISE_SPEED * burn
        damping = balance.INERTIAL_DAMPENING ** (dt * 60.0)
        vx = (float(self.velocity.x) + float(thrust.x) * acceleration * dt) * damping
        vz = (float(self.velocity.y) + float(thrust.y) * acceleration * dt) * damping
        speed = math.hypot(vx, vz)
        if speed > ceiling:
            vx, vz = vx / speed * ceiling, vz / speed * ceiling
        self.velocity = Vec2(vx, vz)
        self.position = Vec3(
            float(self.position.x) + vx * dt,
            PLANE_Y,
            float(self.position.z) + vz * dt,
        )

    def _tick_afterburner(self, dt: float) -> float:
        """Ramp the burn in or out and return the current thrust multiplier."""
        wants = Input.is_action_pressed("afterburner")
        burning = wants and self.drain_energy(balance.ENERGY_AFTERBURNER_PER_S, dt, "afterburner")
        ramp_step = dt / max(balance.AFTERBURNER_IGNITION_RAMP_S, 1e-6)
        if burning:
            self._afterburner_ramp = min(1.0, self._afterburner_ramp + ramp_step)
        else:
            self._afterburner_ramp = max(0.0, self._afterburner_ramp - ramp_step)
        if burning != self._afterburner_active:
            self._afterburner_active = burning
            self.afterburner_changed(burning)
        return 1.0 + (AFTERBURNER_SPEED_MULT - 1.0) * self._afterburner_ramp

    def on_update(self, dt: float):
        self._ensure_camera_rig()
        aim = self.aim_point()
        bearing = bearing_of(
            Vec3(
                float(aim.x) - float(self.position.x),
                0.0,
                float(aim.z) - float(self.position.z),
            )
        )
        if bearing is not None and not self._destroyed:
            self.heading = bearing
            self._aim_direction = heading_to_direction(self.heading)
            self._face_nose()
        if self._rig is not None:
            self._rig.set_aim(aim)
        self._tick_ribbons(dt)
        if self._destroyed:
            return

        self._tick_shield_input(dt)
        self._tick_interact(dt)
        self._tick_channels(dt)

    def _ensure_camera_rig(self) -> None:
        if self._rig is not None or self.tree is None or self.tree.root is None:
            return
        self._rig = _find_camera_rig(self.tree.root)
        if self._rig is not None:
            self._rig.set_target(self)

    def _tick_ribbons(self, dt: float) -> None:
        """Drive the exhaust plumes from the thrust vector and the burner ramp.

        The plume streams opposite the thrust rather than off the tail, because
        the thrust vector and the nose are independent here: a pilot strafing
        sideways with the nose on a target should see which way the ship is
        actually being pushed.

        The throttle the plumes are given is spooled toward the commanded one
        rather than snapped to it, so a tapped key is a jet lighting and dying
        instead of one frame of fire.

        A lit burner floors the throttle at :data:`RIBBON_BURNER_FLOOR` whether
        or not a thrust key is down. It is spending the capacitor either way,
        and a posture the pilot is paying for that shows nothing on the hull is
        a posture they cannot tell they are in.
        """
        if not self._ribbons:
            return
        stalled = self._destroyed or self.patching
        thrust = Vec2(0.0, 0.0) if stalled else self._thrust_input
        burn = 0.0 if stalled else self._afterburner_ramp
        tx, tz = float(thrust.x), float(thrust.y)
        magnitude = min(1.0, math.hypot(tx, tz))
        commanded = magnitude * (RIBBON_CRUISE_THROTTLE + (1.0 - RIBBON_CRUISE_THROTTLE) * burn)
        commanded = max(commanded, RIBBON_BURNER_FLOOR * burn)
        rate = RIBBON_SPOOL_UP_PER_S if commanded > self._ribbon_throttle else RIBBON_SPOOL_DOWN_PER_S
        step = rate * max(dt, 0.0)
        if abs(commanded - self._ribbon_throttle) <= step:
            self._ribbon_throttle = commanded
        else:
            self._ribbon_throttle += math.copysign(step, commanded - self._ribbon_throttle)
        # The plumes are told the drive state only when it moves. Every setter
        # on the ribbon rebuilds the whole plume, and a ship holding a throttle
        # or coasting at zero is the ordinary case, not the exception.
        drive = (self._ribbon_throttle, burn, -tx, -tz)
        if drive == self._ribbon_drive:
            return
        self._ribbon_drive = drive
        exhaust = Vec3(-tx, 0.0, -tz)
        for ribbon in self._ribbons:
            ribbon.set_throttle(self._ribbon_throttle)
            ribbon.set_burn(burn)
            ribbon.set_direction(exhaust)

    def _tick_shield_input(self, dt: float) -> None:
        if self.shield is None:
            return
        left_edge = Input.is_action_just_pressed("shield_left")
        right_edge = Input.is_action_just_pressed("shield_right")
        if self._pad_aim:
            # Bumpers nudge in fixed steps; both together re-centre on the aim.
            step = math.radians(balance.SHIELD_PAD_NUDGE_DEGREES)
            if left_edge and right_edge:
                self.shield.recentre()
            elif left_edge:
                if Input.is_action_pressed("shield_right"):
                    self.shield.recentre()
                else:
                    self.shield.nudge(step)
            elif right_edge:
                if Input.is_action_pressed("shield_left"):
                    self.shield.recentre()
                else:
                    self.shield.nudge(-step)
            return
        rate = SHIELD_ROTATE_RADIANS_PER_S * dt
        if Input.is_action_pressed("shield_left"):
            self.shield.nudge(rate)
        if Input.is_action_pressed("shield_right"):
            self.shield.nudge(-rate)

    def _tick_interact(self, dt: float) -> None:
        """Hold-to-patch, the keyboard and pad route into :meth:`patch_breach`.

        Only the patch rung of the interact ladder lives here: docking, grabbing
        and the Refinery hold belong to the modules that own those verbs.
        """
        if not Input.is_action_pressed("interact"):
            self._interact_hold = 0.0
            self._interact_consumed = False
            return
        self._interact_hold += dt
        if self._interact_consumed or self._interact_hold < INTERACT_PATCH_HOLD_S:
            return
        if self.open_breaches > 0 and not self.patching and self.patch_breach():
            self._interact_consumed = True

    def interact_hold_fraction(self) -> float:
        """Progress of the interact hold toward the patch, 0 to 1."""
        return min(1.0, self._interact_hold / INTERACT_PATCH_HOLD_S)

    def _tick_channels(self, dt: float) -> None:
        if self._patch_remaining > 0.0:
            self._patch_remaining = max(0.0, self._patch_remaining - dt)
            if self._patch_remaining == 0.0 and self.open_breaches > 0:
                self.open_breaches -= 1
                self.breach_patched(self.open_breaches)

        if Input.is_action_just_pressed("warp_spool"):
            if self._spooling:
                self.cancel_warp_spool()
            else:
                self.begin_warp_spool()

        if self._spooling:
            self._spool_remaining -= dt
            if self._spool_remaining <= 0.0:
                if self._fuel_is_dry():
                    self._end_spool(SPOOL_FAILED, SPOOL_FAILURE_NO_FUEL)
                else:
                    self._end_spool(SPOOL_JUMP)


def _find_camera_rig(node: Node) -> CameraRig | None:
    """Depth-first search for the run's camera rig, by type rather than by name."""
    if isinstance(node, CameraRig):
        return node
    for child in node.children:
        found = _find_camera_rig(child)
        if found is not None:
            return found
    return None