feat(perception): add bounded reference graph
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,659 @@
|
||||
"""Persistent bounded reference graph for object-centric recorded-realtime CV."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractContextManager
|
||||
from dataclasses import dataclass, replace
|
||||
from queue import Empty, Full, Queue
|
||||
from threading import Event, Lock, Thread
|
||||
from typing import TypeVar, cast
|
||||
|
||||
from k1link.compute.pipeline_telemetry import (
|
||||
PipelineStageOutcome,
|
||||
PipelineTelemetryEmitter,
|
||||
PipelineTelemetryIdentity,
|
||||
PipelineTelemetrySink,
|
||||
)
|
||||
|
||||
from .baseline import BASELINE_PROFILE_ID, BASELINE_SOURCE_ID
|
||||
from .contracts import (
|
||||
LocalObstacleMap,
|
||||
ObjectProposal2D,
|
||||
ObstacleObservation,
|
||||
SourceAccounting,
|
||||
TemporalObstacle,
|
||||
TemporalState,
|
||||
validate_exclusive_point_ownership,
|
||||
)
|
||||
from .graph_contracts import (
|
||||
GRAPH_RESULT_SCHEMA,
|
||||
REFERENCE_GRAPH_ID,
|
||||
TERMINAL_OUTCOME_SCHEMA,
|
||||
DeliveredFrame,
|
||||
GraphExecutionError,
|
||||
GraphRunResult,
|
||||
GraphState,
|
||||
TerminalOutcome,
|
||||
TerminalOutcomeType,
|
||||
build_graph_run_result,
|
||||
)
|
||||
from .graph_validation import (
|
||||
validate_observations,
|
||||
validate_proposals,
|
||||
validate_temporal,
|
||||
validate_threats,
|
||||
)
|
||||
from .providers import (
|
||||
DetectorProvider,
|
||||
GeometryAssociationProvider,
|
||||
MotionProvider,
|
||||
ProviderRole,
|
||||
ReferencePerceptionGraphConfig,
|
||||
SourcePacket,
|
||||
SourceProvider,
|
||||
TemporalStateProvider,
|
||||
ThreatProvider,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Detected:
|
||||
packet: SourcePacket
|
||||
proposals: tuple[ObjectProposal2D, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Associated:
|
||||
packet: SourcePacket
|
||||
proposals: tuple[ObjectProposal2D, ...]
|
||||
observations: tuple[ObstacleObservation, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Temporal:
|
||||
packet: SourcePacket
|
||||
proposals: tuple[ObjectProposal2D, ...]
|
||||
associated_proposal_ids: frozenset[str]
|
||||
obstacles: tuple[TemporalObstacle, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _StopSignal:
|
||||
pass
|
||||
|
||||
|
||||
_STOP = _StopSignal()
|
||||
_DetectorItem = SourcePacket | _StopSignal
|
||||
_GeometryItem = _Detected | _StopSignal
|
||||
_TemporalItem = _Associated | _StopSignal
|
||||
_ThreatItem = _Temporal | _StopSignal
|
||||
_QueueItem = SourcePacket | _Detected | _Associated | _Temporal | _StopSignal
|
||||
_QueueItemT = TypeVar("_QueueItemT", bound=_QueueItem)
|
||||
|
||||
|
||||
class ReferencePerceptionGraphV1:
|
||||
"""Run one source through bounded provider stages with exact terminal accounting."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
config: ReferencePerceptionGraphConfig,
|
||||
source: SourceProvider,
|
||||
detector: DetectorProvider,
|
||||
geometry: GeometryAssociationProvider,
|
||||
temporal: TemporalStateProvider,
|
||||
motion: MotionProvider,
|
||||
threat: ThreatProvider,
|
||||
telemetry_identity: PipelineTelemetryIdentity | None = None,
|
||||
telemetry_sink: PipelineTelemetrySink | None = None,
|
||||
clock_ns: Callable[[], int] = time.monotonic_ns,
|
||||
) -> None:
|
||||
if config.graph_id != REFERENCE_GRAPH_ID:
|
||||
raise GraphExecutionError("reference graph id is incompatible")
|
||||
if config.source_profile_id != BASELINE_PROFILE_ID:
|
||||
raise GraphExecutionError("reference source profile is incompatible")
|
||||
if (telemetry_identity is None) is not (telemetry_sink is None):
|
||||
raise GraphExecutionError("telemetry identity and sink must be configured together")
|
||||
self.config = config
|
||||
self.source = source
|
||||
self.detector = detector
|
||||
self.geometry = geometry
|
||||
self.temporal = temporal
|
||||
self.motion = motion
|
||||
self.threat = threat
|
||||
self.telemetry_identity = telemetry_identity
|
||||
self.telemetry_sink = telemetry_sink
|
||||
self._clock_ns = clock_ns
|
||||
self._state = GraphState.CREATED
|
||||
self._state_lock = Lock()
|
||||
self._result_lock = Lock()
|
||||
self._cancel_event = Event()
|
||||
self._threads: list[Thread] = []
|
||||
self._outcomes: dict[int, TerminalOutcome] = {}
|
||||
self._deliveries: list[DeliveredFrame] = []
|
||||
self._admitted: dict[int, SourcePacket] = {}
|
||||
self._admitted_at_ns: dict[int, int] = {}
|
||||
self._validate_provider_pins()
|
||||
|
||||
@property
|
||||
def state(self) -> GraphState:
|
||||
with self._state_lock:
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def worker_threads_alive(self) -> int:
|
||||
return sum(thread.is_alive() for thread in self._threads)
|
||||
|
||||
def cancel(self) -> None:
|
||||
self._cancel_event.set()
|
||||
|
||||
def run(self) -> GraphRunResult:
|
||||
self._begin_run()
|
||||
started_ns = self._now()
|
||||
detector_queue: Queue[_DetectorItem] = Queue(self._capacity("detector"))
|
||||
geometry_queue: Queue[_GeometryItem] = Queue(self._capacity("geometry"))
|
||||
temporal_queue: Queue[_TemporalItem] = Queue(self._capacity("temporal"))
|
||||
threat_queue: Queue[_ThreatItem] = Queue(self._capacity("threat"))
|
||||
self._threads = [
|
||||
Thread(
|
||||
target=self._detector_loop,
|
||||
args=(detector_queue, geometry_queue),
|
||||
name="m4-detector",
|
||||
daemon=True,
|
||||
),
|
||||
Thread(
|
||||
target=self._geometry_loop,
|
||||
args=(geometry_queue, temporal_queue),
|
||||
name="m4-geometry",
|
||||
daemon=True,
|
||||
),
|
||||
Thread(
|
||||
target=self._temporal_loop,
|
||||
args=(temporal_queue, threat_queue),
|
||||
name="m4-temporal",
|
||||
daemon=True,
|
||||
),
|
||||
Thread(
|
||||
target=self._threat_loop,
|
||||
args=(threat_queue,),
|
||||
name="m4-threat",
|
||||
daemon=True,
|
||||
),
|
||||
]
|
||||
for thread in self._threads:
|
||||
thread.start()
|
||||
run_emitter = self._run_emitter()
|
||||
if run_emitter is not None:
|
||||
run_emitter.run("started")
|
||||
self._set_state(GraphState.RUNNING)
|
||||
source_failed = False
|
||||
try:
|
||||
for packet in self.source.packets(self._cancel_event):
|
||||
if self._cancel_event.is_set():
|
||||
break
|
||||
if not self._admit(packet):
|
||||
continue
|
||||
self._put_latest(detector_queue, packet, "detector")
|
||||
except Exception:
|
||||
source_failed = True
|
||||
finally:
|
||||
self._set_state(GraphState.STOPPING)
|
||||
self._put_stop(detector_queue, "detector")
|
||||
|
||||
self._join_workers()
|
||||
self._close_accounting()
|
||||
final_state = (
|
||||
GraphState.CANCELLED
|
||||
if self._cancel_event.is_set()
|
||||
else GraphState.FAILED
|
||||
if source_failed
|
||||
else GraphState.STOPPED
|
||||
)
|
||||
self._set_state(final_state)
|
||||
result = self._result(final_state)
|
||||
if run_emitter is not None:
|
||||
duration_ms = max(0.0, (self._now() - started_ns) / 1_000_000)
|
||||
if final_state is GraphState.STOPPED:
|
||||
run_emitter.run("completed", duration_ms=duration_ms, exit_code=0)
|
||||
else:
|
||||
run_emitter.run(
|
||||
"failed",
|
||||
duration_ms=duration_ms,
|
||||
exit_code=1,
|
||||
error_type=(
|
||||
"GraphCancelled" if final_state is GraphState.CANCELLED else "SourceFailure"
|
||||
),
|
||||
)
|
||||
return result
|
||||
|
||||
def _begin_run(self) -> None:
|
||||
with self._state_lock:
|
||||
if self._state not in {
|
||||
GraphState.CREATED,
|
||||
GraphState.STOPPED,
|
||||
GraphState.CANCELLED,
|
||||
GraphState.FAILED,
|
||||
}:
|
||||
raise GraphExecutionError("reference graph is already running")
|
||||
self._state = GraphState.STARTING
|
||||
self._cancel_event = Event()
|
||||
with self._result_lock:
|
||||
self._outcomes.clear()
|
||||
self._deliveries.clear()
|
||||
self._admitted.clear()
|
||||
self._admitted_at_ns.clear()
|
||||
|
||||
def _admit(self, packet: SourcePacket) -> bool:
|
||||
envelope = packet.envelope
|
||||
with self._result_lock:
|
||||
if envelope.sequence in self._admitted:
|
||||
return False
|
||||
self._admitted[envelope.sequence] = packet
|
||||
self._admitted_at_ns[envelope.sequence] = self._now()
|
||||
if envelope.source_id != BASELINE_SOURCE_ID:
|
||||
self._terminal(packet, TerminalOutcomeType.REJECTED, "source", "source-not-admitted")
|
||||
return False
|
||||
if not envelope.image.available:
|
||||
self._terminal(packet, TerminalOutcomeType.UNAVAILABLE, "source", "image-unavailable")
|
||||
return False
|
||||
if envelope.source_age_ns > self._deadline("detector"):
|
||||
self._terminal(packet, TerminalOutcomeType.STALE, "source", "source-deadline-exceeded")
|
||||
return False
|
||||
return True
|
||||
|
||||
def _detector_loop(
|
||||
self,
|
||||
incoming: Queue[_DetectorItem],
|
||||
outgoing: Queue[_GeometryItem],
|
||||
) -> None:
|
||||
while True:
|
||||
item = incoming.get()
|
||||
try:
|
||||
if isinstance(item, _StopSignal):
|
||||
self._put_stop(outgoing, "geometry")
|
||||
return
|
||||
if self._cancelled(item):
|
||||
continue
|
||||
if self._expired(item, "detector"):
|
||||
continue
|
||||
with self._stage(item, "detector", 1) as stage:
|
||||
proposals = self.detector.detect(item)
|
||||
validate_proposals(item, proposals)
|
||||
if stage is not None:
|
||||
stage.output_count = len(proposals)
|
||||
self._put_latest(outgoing, _Detected(item, proposals), "geometry")
|
||||
except Exception as exc:
|
||||
if not isinstance(item, _StopSignal):
|
||||
self._failed(item, "detector", exc)
|
||||
finally:
|
||||
incoming.task_done()
|
||||
|
||||
def _geometry_loop(
|
||||
self,
|
||||
incoming: Queue[_GeometryItem],
|
||||
outgoing: Queue[_TemporalItem],
|
||||
) -> None:
|
||||
while True:
|
||||
item = incoming.get()
|
||||
try:
|
||||
if isinstance(item, _StopSignal):
|
||||
self._put_stop(outgoing, "temporal")
|
||||
return
|
||||
if self._cancelled(item.packet):
|
||||
continue
|
||||
if self._expired(item.packet, "geometry"):
|
||||
continue
|
||||
with self._stage(item.packet, "geometry", len(item.proposals)) as stage:
|
||||
observations = self.geometry.associate(item.packet, item.proposals)
|
||||
validate_observations(item.packet, item.proposals, observations)
|
||||
validate_exclusive_point_ownership(observations)
|
||||
if stage is not None:
|
||||
stage.output_count = len(observations)
|
||||
self._put_latest(
|
||||
outgoing,
|
||||
_Associated(item.packet, item.proposals, observations),
|
||||
"temporal",
|
||||
)
|
||||
except Exception as exc:
|
||||
if not isinstance(item, _StopSignal):
|
||||
self._failed(item.packet, "geometry", exc)
|
||||
finally:
|
||||
incoming.task_done()
|
||||
|
||||
def _temporal_loop(
|
||||
self,
|
||||
incoming: Queue[_TemporalItem],
|
||||
outgoing: Queue[_ThreatItem],
|
||||
) -> None:
|
||||
while True:
|
||||
item = incoming.get()
|
||||
try:
|
||||
if isinstance(item, _StopSignal):
|
||||
self._put_stop(outgoing, "threat")
|
||||
return
|
||||
if self._cancelled(item.packet):
|
||||
continue
|
||||
if self._expired(item.packet, "temporal"):
|
||||
continue
|
||||
with self._stage(item.packet, "temporal", len(item.observations)) as stage:
|
||||
obstacles = self.temporal.update(item.packet, item.observations)
|
||||
obstacles = self.motion.estimate(item.packet, obstacles)
|
||||
validate_temporal(item.packet, obstacles)
|
||||
if stage is not None:
|
||||
stage.output_count = len(obstacles)
|
||||
self._put_latest(
|
||||
outgoing,
|
||||
_Temporal(
|
||||
item.packet,
|
||||
item.proposals,
|
||||
frozenset(
|
||||
proposal_id
|
||||
for observation in item.observations
|
||||
if observation.occupied_support
|
||||
for proposal_id in observation.proposal_ids
|
||||
),
|
||||
obstacles,
|
||||
),
|
||||
"threat",
|
||||
)
|
||||
except Exception as exc:
|
||||
if not isinstance(item, _StopSignal):
|
||||
self._failed(item.packet, "temporal", exc)
|
||||
finally:
|
||||
incoming.task_done()
|
||||
|
||||
def _threat_loop(self, incoming: Queue[_ThreatItem]) -> None:
|
||||
while True:
|
||||
item = incoming.get()
|
||||
try:
|
||||
if isinstance(item, _StopSignal):
|
||||
return
|
||||
if self._cancelled(item.packet):
|
||||
continue
|
||||
if self._expired(item.packet, "threat"):
|
||||
continue
|
||||
obstacle_map = self._obstacle_map(item)
|
||||
with self._stage(item.packet, "threat", len(item.obstacles)) as stage:
|
||||
threats = self.threat.assess(obstacle_map)
|
||||
validate_threats(obstacle_map, threats)
|
||||
if stage is not None:
|
||||
stage.output_count = len(threats)
|
||||
delivery = DeliveredFrame(
|
||||
sequence=item.packet.envelope.sequence,
|
||||
obstacle_map=obstacle_map,
|
||||
threats=threats,
|
||||
)
|
||||
with self._result_lock:
|
||||
self._deliveries.append(delivery)
|
||||
self._terminal(
|
||||
item.packet,
|
||||
TerminalOutcomeType.DELIVERED,
|
||||
"threat",
|
||||
"object-payload-delivered",
|
||||
)
|
||||
except Exception as exc:
|
||||
if not isinstance(item, _StopSignal):
|
||||
self._failed(item.packet, "threat", exc)
|
||||
finally:
|
||||
incoming.task_done()
|
||||
|
||||
def _obstacle_map(self, item: _Temporal) -> LocalObstacleMap:
|
||||
packet = item.packet
|
||||
now_ns = self._now()
|
||||
with self._result_lock:
|
||||
admitted_at_ns = self._admitted_at_ns[packet.envelope.sequence]
|
||||
occupied = tuple(
|
||||
obstacle for obstacle in item.obstacles if obstacle.state is TemporalState.CURRENT
|
||||
)
|
||||
unknown = tuple(
|
||||
obstacle for obstacle in item.obstacles if obstacle.state is not TemporalState.CURRENT
|
||||
)
|
||||
camera_uncertainty = tuple(
|
||||
proposal
|
||||
for proposal in item.proposals
|
||||
if proposal.proposal_id not in item.associated_proposal_ids
|
||||
)
|
||||
return LocalObstacleMap(
|
||||
source_id=packet.envelope.source_id,
|
||||
session_id=packet.envelope.session_id,
|
||||
frame_id=packet.envelope.frame_id,
|
||||
graph_id=self.config.graph_id,
|
||||
generated_monotonic_ns=now_ns,
|
||||
output_age_ns=(
|
||||
packet.envelope.source_age_ns + max(0, now_ns - admitted_at_ns)
|
||||
),
|
||||
occupied=occupied,
|
||||
unknown=unknown,
|
||||
camera_uncertainty=camera_uncertainty,
|
||||
accounting=SourceAccounting(1, 1, 0, 0),
|
||||
)
|
||||
|
||||
def _cancelled(self, packet: SourcePacket) -> bool:
|
||||
if not self._cancel_event.is_set():
|
||||
return False
|
||||
self._terminal(packet, TerminalOutcomeType.FAILED, "graph", "graph-cancelled")
|
||||
return True
|
||||
|
||||
def _expired(self, packet: SourcePacket, stage_id: str) -> bool:
|
||||
with self._result_lock:
|
||||
admitted_at_ns = self._admitted_at_ns[packet.envelope.sequence]
|
||||
execution_age_ns = max(0, self._now() - admitted_at_ns)
|
||||
if packet.envelope.source_age_ns + execution_age_ns <= self._deadline(stage_id):
|
||||
return False
|
||||
self._terminal(
|
||||
packet,
|
||||
TerminalOutcomeType.STALE,
|
||||
stage_id,
|
||||
"stage-deadline-exceeded",
|
||||
)
|
||||
return True
|
||||
|
||||
def _failed(self, packet: SourcePacket, stage_id: str, exc: Exception) -> None:
|
||||
self._terminal(
|
||||
packet,
|
||||
TerminalOutcomeType.FAILED,
|
||||
stage_id,
|
||||
f"provider-error-{type(exc).__name__.lower()}",
|
||||
)
|
||||
|
||||
def _terminal(
|
||||
self,
|
||||
packet: SourcePacket,
|
||||
outcome: TerminalOutcomeType,
|
||||
stage_id: str,
|
||||
reason: str,
|
||||
) -> None:
|
||||
envelope = packet.envelope
|
||||
terminal = TerminalOutcome(
|
||||
source_id=envelope.source_id,
|
||||
session_id=envelope.session_id,
|
||||
frame_id=envelope.frame_id,
|
||||
sequence=envelope.sequence,
|
||||
outcome=outcome,
|
||||
stage_id=stage_id,
|
||||
reason=reason,
|
||||
)
|
||||
with self._result_lock:
|
||||
self._outcomes.setdefault(envelope.sequence, terminal)
|
||||
|
||||
def _put_latest(
|
||||
self,
|
||||
queue: Queue[_QueueItemT],
|
||||
item: _QueueItemT,
|
||||
stage_id: str,
|
||||
) -> None:
|
||||
while True:
|
||||
try:
|
||||
queue.put_nowait(item)
|
||||
return
|
||||
except Full:
|
||||
pass
|
||||
try:
|
||||
dropped = queue.get_nowait()
|
||||
except Empty:
|
||||
continue
|
||||
try:
|
||||
if not isinstance(dropped, _StopSignal):
|
||||
packet = _packet_from_item(dropped)
|
||||
self._terminal(
|
||||
packet,
|
||||
TerminalOutcomeType.SUPERSEDED,
|
||||
stage_id,
|
||||
"bounded-queue-latest-wins",
|
||||
)
|
||||
finally:
|
||||
queue.task_done()
|
||||
|
||||
def _put_stop(self, queue: Queue[_QueueItemT], stage_id: str) -> None:
|
||||
timeout = self._terminal_timeout(stage_id) / 1_000_000_000
|
||||
try:
|
||||
queue.put(cast(_QueueItemT, _STOP), timeout=timeout)
|
||||
except Full:
|
||||
self._put_latest(queue, cast(_QueueItemT, _STOP), stage_id)
|
||||
|
||||
def _join_workers(self) -> None:
|
||||
timeout_seconds = max(
|
||||
1.0,
|
||||
max(policy.terminal_timeout_ns for policy in self.config.queues)
|
||||
/ 1_000_000_000
|
||||
* 2,
|
||||
)
|
||||
for thread in self._threads:
|
||||
thread.join(timeout_seconds)
|
||||
if self.worker_threads_alive:
|
||||
self._set_state(GraphState.FAILED)
|
||||
raise GraphExecutionError("reference graph left an orphan worker thread")
|
||||
|
||||
def _close_accounting(self) -> None:
|
||||
with self._result_lock:
|
||||
missing = sorted(set(self._admitted) - set(self._outcomes))
|
||||
packets = [self._admitted[sequence] for sequence in missing]
|
||||
for packet in packets:
|
||||
self._terminal(
|
||||
packet,
|
||||
TerminalOutcomeType.FAILED,
|
||||
"graph",
|
||||
"terminal-accounting-gap",
|
||||
)
|
||||
with self._result_lock:
|
||||
if set(self._admitted) != set(self._outcomes):
|
||||
raise GraphExecutionError("terminal accounting did not close")
|
||||
|
||||
def _result(self, state: GraphState) -> GraphRunResult:
|
||||
with self._result_lock:
|
||||
outcomes = tuple(self._outcomes[key] for key in sorted(self._outcomes))
|
||||
deliveries = tuple(sorted(self._deliveries, key=lambda item: item.sequence))
|
||||
admitted_count = len(self._admitted)
|
||||
return build_graph_run_result(
|
||||
graph_id=self.config.graph_id,
|
||||
source_profile_id=self.config.source_profile_id,
|
||||
state=state,
|
||||
admitted_count=admitted_count,
|
||||
outcomes=outcomes,
|
||||
deliveries=deliveries,
|
||||
)
|
||||
|
||||
def _stage(
|
||||
self,
|
||||
packet: SourcePacket,
|
||||
stage_id: str,
|
||||
input_count: int,
|
||||
) -> AbstractContextManager[PipelineStageOutcome | _NullStage]:
|
||||
emitter = self._frame_emitter(packet)
|
||||
if emitter is None:
|
||||
return _NullStage()
|
||||
return emitter.stage(stage_id, input_count=input_count)
|
||||
|
||||
def _run_emitter(self) -> PipelineTelemetryEmitter | None:
|
||||
if self.telemetry_identity is None or self.telemetry_sink is None:
|
||||
return None
|
||||
return PipelineTelemetryEmitter(
|
||||
identity=self.telemetry_identity,
|
||||
sink=self.telemetry_sink,
|
||||
clock_ns=self._clock_ns,
|
||||
)
|
||||
|
||||
def _frame_emitter(self, packet: SourcePacket) -> PipelineTelemetryEmitter | None:
|
||||
if self.telemetry_identity is None or self.telemetry_sink is None:
|
||||
return None
|
||||
identity = replace(
|
||||
self.telemetry_identity,
|
||||
request_id=packet.envelope.frame_id,
|
||||
frame_index=packet.envelope.sequence,
|
||||
)
|
||||
return PipelineTelemetryEmitter(
|
||||
identity=identity,
|
||||
sink=self.telemetry_sink,
|
||||
clock_ns=self._clock_ns,
|
||||
)
|
||||
|
||||
def _validate_provider_pins(self) -> None:
|
||||
actual = {
|
||||
ProviderRole.SOURCE: self.source.provider_id,
|
||||
ProviderRole.DETECTOR: self.detector.provider_id,
|
||||
ProviderRole.GEOMETRY: self.geometry.provider_id,
|
||||
ProviderRole.TEMPORAL: self.temporal.provider_id,
|
||||
ProviderRole.MOTION: self.motion.provider_id,
|
||||
ProviderRole.THREAT: self.threat.provider_id,
|
||||
}
|
||||
pinned = {pin.role: pin.provider_id for pin in self.config.providers}
|
||||
if actual != pinned:
|
||||
raise GraphExecutionError("configured provider identities do not match runtime")
|
||||
|
||||
def _capacity(self, stage_id: str) -> int:
|
||||
return next(policy.capacity for policy in self.config.queues if policy.stage_id == stage_id)
|
||||
|
||||
def _deadline(self, stage_id: str) -> int:
|
||||
return next(
|
||||
policy.deadline_ns
|
||||
for policy in self.config.queues
|
||||
if policy.stage_id == stage_id
|
||||
)
|
||||
|
||||
def _terminal_timeout(self, stage_id: str) -> int:
|
||||
return next(
|
||||
policy.terminal_timeout_ns
|
||||
for policy in self.config.queues
|
||||
if policy.stage_id == stage_id
|
||||
)
|
||||
|
||||
def _now(self) -> int:
|
||||
return int(self._clock_ns())
|
||||
|
||||
def _set_state(self, state: GraphState) -> None:
|
||||
with self._state_lock:
|
||||
self._state = state
|
||||
|
||||
|
||||
class _NullStage:
|
||||
output_count: int | None = None
|
||||
|
||||
def __enter__(self) -> _NullStage:
|
||||
return self
|
||||
|
||||
def __exit__(self, *args: object) -> None:
|
||||
return None
|
||||
|
||||
|
||||
def _packet_from_item(item: object) -> SourcePacket:
|
||||
if isinstance(item, SourcePacket):
|
||||
return item
|
||||
if isinstance(item, (_Detected, _Associated, _Temporal)):
|
||||
return item.packet
|
||||
raise GraphExecutionError("queue contained an incompatible item")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"GRAPH_RESULT_SCHEMA",
|
||||
"REFERENCE_GRAPH_ID",
|
||||
"TERMINAL_OUTCOME_SCHEMA",
|
||||
"DeliveredFrame",
|
||||
"GraphExecutionError",
|
||||
"GraphRunResult",
|
||||
"GraphState",
|
||||
"ReferencePerceptionGraphV1",
|
||||
"TerminalOutcome",
|
||||
"TerminalOutcomeType",
|
||||
]
|
||||
Reference in New Issue
Block a user