diff --git a/apps/control-station/src/components/system/WorkerPipelineStages.tsx b/apps/control-station/src/components/system/WorkerPipelineStages.tsx
new file mode 100644
index 0000000..a6cf3ee
--- /dev/null
+++ b/apps/control-station/src/components/system/WorkerPipelineStages.tsx
@@ -0,0 +1,72 @@
+import type { CSSProperties } from "react";
+
+import type { WorkerPipelineStage } from "../../core/system/workerTelemetry";
+
+interface WorkerPipelineStagesProps {
+ stages: WorkerPipelineStage[];
+}
+
+function stateLabel(state: WorkerPipelineStage["state"]): string {
+ if (state === "active") return "выполняется";
+ if (state === "waiting") return "ожидает";
+ if (state === "ready") return "готов";
+ return "нет live-состояния";
+}
+
+function stageDuration(value: number | null): string {
+ if (typeof value !== "number" || !Number.isFinite(value)) return "—";
+ if (value < 1) return `${Math.round(value * 1000)} мс`;
+ return `${new Intl.NumberFormat("ru-RU", {
+ maximumFractionDigits: value < 10 ? 2 : 1,
+ }).format(value)} с`;
+}
+
+function stageDetail(stage: WorkerPipelineStage): string {
+ if (
+ typeof stage.activations !== "number"
+ || typeof stage.share_percent !== "number"
+ || !Number.isFinite(stage.share_percent)
+ ) {
+ return "Измерений ещё нет";
+ }
+ const share = new Intl.NumberFormat("ru-RU", {
+ maximumFractionDigits: 1,
+ }).format(stage.share_percent);
+ return `${stage.activations} проходов · ${share}% измеренного времени`;
+}
+
+export function WorkerPipelineStages({ stages }: WorkerPipelineStagesProps) {
+ return (
+
+ {stages.map((stage, index) => {
+ const share = typeof stage.share_percent === "number"
+ ? Math.max(0, Math.min(100, stage.share_percent))
+ : 0;
+ const style = {
+ "--worker-stage-share": `${share}%`,
+ } as CSSProperties;
+ return (
+ -
+
+
+ {String(index + 1).padStart(2, "0")}
+
+
+ {stage.label}
+ {stageDetail(stage)}
+
+
+ {stageDuration(stage.elapsed_seconds)}
+ {stateLabel(stage.state)}
+
+
+ );
+ })}
+
+ );
+}
diff --git a/apps/control-station/src/core/system/workerTelemetry.ts b/apps/control-station/src/core/system/workerTelemetry.ts
index 326663b..b9a70f8 100644
--- a/apps/control-station/src/core/system/workerTelemetry.ts
+++ b/apps/control-station/src/core/system/workerTelemetry.ts
@@ -51,6 +51,9 @@ export interface WorkerPipelineStage {
id: string;
label: string;
state: "active" | "waiting" | "ready" | "unavailable";
+ elapsed_seconds: number | null;
+ share_percent: number | null;
+ activations: number | null;
}
export interface WorkerTelemetryHistoryRow {
diff --git a/apps/control-station/src/styles/system-telemetry.css b/apps/control-station/src/styles/system-telemetry.css
index 8966380..71c164a 100644
--- a/apps/control-station/src/styles/system-telemetry.css
+++ b/apps/control-station/src/styles/system-telemetry.css
@@ -299,7 +299,7 @@
.worker-stage-list {
display: grid;
min-width: 0;
- grid-template-columns: repeat(3, minmax(0, 1fr));
+ grid-template-columns: minmax(0, 1fr);
margin: 0;
padding: 0;
gap: 0.48rem;
@@ -307,12 +307,15 @@
}
.worker-stage-list li {
+ position: relative;
display: grid;
+ overflow: hidden;
min-width: 0;
- grid-template-columns: auto minmax(0, 1fr) auto;
+ grid-template-columns: 2rem minmax(0, 1fr) minmax(7.5rem, auto);
align-items: center;
- gap: 0.55rem;
- padding: 0.66rem;
+ gap: 0.75rem;
+ padding: 0.72rem 0.8rem;
+ isolation: isolate;
border: 1px solid var(--station-hairline);
border-radius: var(--nodedc-radius-control);
background: var(--station-panel-soft);
@@ -323,21 +326,53 @@
background: var(--nodedc-focus-surface);
}
-.worker-stage-list li > span,
-.worker-stage-list li > small {
+.worker-stage-list__fill {
+ position: absolute;
+ z-index: -1;
+ inset: 0 auto 0 0;
+ width: var(--worker-stage-share, 0%);
+ background: color-mix(
+ in srgb,
+ var(--nodedc-text-primary) 9%,
+ transparent
+ );
+ pointer-events: none;
+}
+
+.worker-stage-list li[data-state="active"] .worker-stage-list__fill {
+ background: color-mix(
+ in srgb,
+ var(--nodedc-text-primary) 15%,
+ transparent
+ );
+}
+
+.worker-stage-list__index,
+.worker-stage-list small {
color: var(--nodedc-text-muted);
font-size: 0.53rem;
}
-.worker-stage-list li > strong {
+.worker-stage-list li > div {
+ display: grid;
+ min-width: 0;
+ gap: 0.2rem;
+}
+
+.worker-stage-list li strong {
min-width: 0;
overflow: hidden;
color: var(--nodedc-text-primary);
- font-size: 0.62rem;
+ font-size: 0.65rem;
text-overflow: ellipsis;
white-space: nowrap;
}
+.worker-stage-list__value {
+ justify-items: end;
+ text-align: right;
+}
+
.network-stat-card {
display: grid;
align-content: center;
@@ -483,7 +518,6 @@
grid-template-columns: repeat(2, minmax(0, 1fr));
}
- .worker-stage-list,
.network-interface-list {
grid-template-columns: repeat(2, minmax(0, 1fr));
}
@@ -522,12 +556,21 @@
.network-overview-grid,
.worker-hardware__facts,
.worker-runtime-grid,
- .worker-stage-list,
.network-interface-list,
.network-profile__security {
grid-template-columns: 1fr;
}
+ .worker-stage-list li {
+ grid-template-columns: 1.5rem minmax(0, 1fr);
+ }
+
+ .worker-stage-list__value {
+ grid-column: 2;
+ justify-items: start;
+ text-align: left;
+ }
+
.worker-runtime-card dl,
.worker-pipeline__summary {
grid-template-columns: repeat(2, minmax(0, 1fr));
diff --git a/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx b/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx
index 93c5f05..8616bd8 100644
--- a/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx
+++ b/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx
@@ -6,6 +6,7 @@ import {
} from "@nodedc/ui-react";
import { TelemetrySeries } from "../../components/system/TelemetrySeries";
+import { WorkerPipelineStages } from "../../components/system/WorkerPipelineStages";
import { WorkerRuntimeCard } from "../../components/system/WorkerRuntimeCard";
import {
formatBytes,
@@ -189,20 +190,7 @@ export function ComputeModulesWorkspace() {
Inference success{node?.triton.requests_succeeded ?? "—"}
Inference failed{node?.triton.requests_failed ?? "—"}
-
- {(telemetry?.pipeline.stages ?? []).map((stage, index) => (
- -
- {String(index + 1).padStart(2, "0")}
- {stage.label}
- {
- stage.state === "active" ? "выполняется"
- : stage.state === "waiting" ? "в очереди"
- : stage.state === "ready" ? "готов"
- : "нет данных"
- }
-
- ))}
-
+
diff --git a/apps/control-station/test/systemTelemetry.test.mjs b/apps/control-station/test/systemTelemetry.test.mjs
index b8ee1c3..aafeeb3 100644
--- a/apps/control-station/test/systemTelemetry.test.mjs
+++ b/apps/control-station/test/systemTelemetry.test.mjs
@@ -14,6 +14,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
workspaceHub,
core,
computeWorkspace,
+ pipelineStages,
networkWorkspace,
styles,
] = await Promise.all([
@@ -21,6 +22,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
read("workspaces/Workspaces.tsx"),
read("core/system/workerTelemetry.ts"),
read("workspaces/system/ComputeModulesWorkspace.tsx"),
+ read("components/system/WorkerPipelineStages.tsx"),
read("workspaces/system/NetworkWorkspace.tsx"),
read("styles.css"),
]);
@@ -32,8 +34,12 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
assert.doesNotMatch(workspaceHub, /worker-telemetry|worker-profile|DESKTOP-OPJ8J04/);
assert.match(core, /\/api\/v1\/system\/worker-telemetry/);
assert.match(core, /\/api\/v1\/system\/worker-profile/);
+ assert.match(core, /share_percent/);
assert.doesNotMatch(core, /@nodedc\/ui-react/);
assert.match(computeWorkspace, /useWorkerTelemetry/);
+ assert.match(computeWorkspace, /WorkerPipelineStages/);
+ assert.match(pipelineStages, /измеренного времени/);
+ assert.doesNotMatch(pipelineStages, /CPU|GPU|hardware/);
assert.match(networkWorkspace, /127\.0\.0\.1:8000/);
assert.doesNotMatch(networkWorkspace, /8765/);
assert.match(styles, /system-telemetry\.css/);
diff --git a/experiments/perception/worker/e15_shadow_runtime.py b/experiments/perception/worker/e15_shadow_runtime.py
index 3d3606b..2307dfa 100644
--- a/experiments/perception/worker/e15_shadow_runtime.py
+++ b/experiments/perception/worker/e15_shadow_runtime.py
@@ -6,6 +6,7 @@ import threading
import time
from collections import deque
from collections.abc import Callable
+from contextlib import AbstractContextManager, nullcontext
from dataclasses import dataclass
from typing import Any
@@ -168,6 +169,7 @@ class PersistentFmp4Decoder:
height: int = 600,
maximum_buffer_bytes: int = 8 * 1024 * 1024,
metadata_capacity: int = 16,
+ measure_decode: Callable[[int], AbstractContextManager[None]] | None = None,
) -> None:
if width < 1 or height < 1:
raise ValueError("decoder resolution is invalid")
@@ -176,6 +178,7 @@ class PersistentFmp4Decoder:
self._height = height
self._media = IncrementalMediaBuffer(maximum_buffer_bytes)
self._metadata = CameraMetadataQueue(metadata_capacity)
+ self._measure_decode = measure_decode
self._thread = threading.Thread(
target=self._decode,
name="lab-e15-fmp4-decoder",
@@ -262,27 +265,33 @@ class PersistentFmp4Decoder:
try:
for frame in container.decode(video=0):
metadata = self._metadata.take()
- image = frame.to_ndarray(format="rgb24")
- if image.shape != (self._height, self._width, 3):
- raise ShadowRuntimeError("decoded camera resolution changed")
- image.setflags(write=False)
- decoded_monotonic = time.perf_counter()
- self._on_frame(
- DecodedCameraFrame(
- frame_index=self._decoded_frames,
- metadata=metadata,
- image=image,
- decoded_monotonic=decoded_monotonic,
- decode_age_ms=max(
- 0.0,
- (
- decoded_monotonic
- - metadata.worker_received_monotonic
- )
- * 1000,
- ),
- )
+ measurement = (
+ self._measure_decode(self._decoded_frames)
+ if self._measure_decode is not None
+ else nullcontext()
)
+ with measurement:
+ image = frame.to_ndarray(format="rgb24")
+ if image.shape != (self._height, self._width, 3):
+ raise ShadowRuntimeError("decoded camera resolution changed")
+ image.setflags(write=False)
+ decoded_monotonic = time.perf_counter()
+ self._on_frame(
+ DecodedCameraFrame(
+ frame_index=self._decoded_frames,
+ metadata=metadata,
+ image=image,
+ decoded_monotonic=decoded_monotonic,
+ decode_age_ms=max(
+ 0.0,
+ (
+ decoded_monotonic
+ - metadata.worker_received_monotonic
+ )
+ * 1000,
+ ),
+ )
+ )
self._decoded_frames += 1
finally:
container.close()
diff --git a/experiments/perception/worker/run_e15_shadow_inference.py b/experiments/perception/worker/run_e15_shadow_inference.py
index 72c5c17..dc34afe 100644
--- a/experiments/perception/worker/run_e15_shadow_inference.py
+++ b/experiments/perception/worker/run_e15_shadow_inference.py
@@ -20,7 +20,8 @@ import sys
import threading
import time
from collections import Counter, deque
-from contextlib import suppress
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
@@ -115,6 +116,17 @@ FUSION_SCHEMA = "missioncore.e15-shadow-fusion-frame/v1"
WORLD_SCHEMA = "missioncore.live-perception-world-state/v1"
SEMANTIC_SCHEMA = "missioncore.e15-shadow-semantic-frame/v1"
PIPELINE_ID = "shadow-fmp4-yolox-eomt-kb4-amodal-world-state/v1"
+PIPELINE_STAGE_IDS = (
+ "source-ingress",
+ "camera-decode",
+ "preprocessing",
+ "detector",
+ "semantic-model",
+ "sensor-fusion",
+ "tracking",
+ "temporal-state",
+ "result-publication",
+)
def arguments() -> argparse.Namespace:
@@ -778,6 +790,76 @@ class _RuntimeTelemetry:
return summary
+class _StageExecutionTelemetry:
+ """Measure named pipeline spans without pretending they are OS processes."""
+
+ def __init__(self, stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS) -> None:
+ if not stage_ids or len(stage_ids) != len(set(stage_ids)):
+ raise RuntimeError("pipeline stage identities are invalid")
+ self._stage_ids = stage_ids
+ self._lock = threading.Lock()
+ self._next_token = 0
+ self._active: dict[int, tuple[str, float, int | None]] = {}
+ self._elapsed_seconds = dict.fromkeys(stage_ids, 0.0)
+ self._activations = dict.fromkeys(stage_ids, 0)
+ self._last_frame_index: int | None = None
+
+ @contextmanager
+ def measure(
+ self,
+ stage_id: str,
+ frame_index: int | None = None,
+ ) -> Iterator[None]:
+ if stage_id not in self._elapsed_seconds:
+ raise RuntimeError(f"unknown pipeline stage: {stage_id}")
+ started = time.perf_counter()
+ with self._lock:
+ self._next_token += 1
+ token = self._next_token
+ self._active[token] = (stage_id, started, frame_index)
+ self._activations[stage_id] += 1
+ if frame_index is not None:
+ self._last_frame_index = frame_index
+ try:
+ yield
+ finally:
+ finished = time.perf_counter()
+ with self._lock:
+ active = self._active.pop(token, None)
+ if active is not None:
+ self._elapsed_seconds[stage_id] += max(0.0, finished - active[1])
+
+ def snapshot(self) -> dict[str, Any]:
+ now = time.perf_counter()
+ with self._lock:
+ elapsed = dict(self._elapsed_seconds)
+ active_rows = list(self._active.values())
+ for stage_id, started, _frame_index in active_rows:
+ elapsed[stage_id] += max(0.0, now - started)
+ total = sum(elapsed.values())
+ active_stages = list(
+ dict.fromkeys(stage_id for stage_id, _started, _frame in active_rows)
+ )
+ current_stage = active_rows[-1][0] if active_rows else None
+ return {
+ "current_stage": current_stage,
+ "active_stages": active_stages,
+ "active_frame_index": self._last_frame_index,
+ "stages": {
+ stage_id: {
+ "elapsed_seconds": round(elapsed[stage_id], 6),
+ "activations": self._activations[stage_id],
+ "share_percent": (
+ round(elapsed[stage_id] / total * 100, 6)
+ if total > 0
+ else None
+ ),
+ }
+ for stage_id in self._stage_ids
+ },
+ }
+
+
def _send_client_binary_frame(stream: Any, payload: bytes) -> None:
if not payload or len(payload) > LIVE_RESULT_MAX_PAYLOAD_BYTES + 256 * 1024 + 8:
raise ShadowRuntimeError("live result websocket frame exceeds the bound")
@@ -810,6 +892,7 @@ def _receiver(
sensor_decode_ms: dict[str, list[float]],
result_queue: queue.Queue[bytes],
result_complete: threading.Event,
+ stage_telemetry: _StageExecutionTelemetry,
) -> None:
import select
@@ -871,7 +954,8 @@ def _receiver(
continue
if opcode != 0x2:
raise ShadowRuntimeError(f"unexpected websocket opcode: {opcode}")
- header, payload = _decode_event(frame)
+ with stage_telemetry.measure("source-ingress"):
+ header, payload = _decode_event(frame)
sequence = int(header["ingress_sequence"])
if state.last_ingress_sequence is not None:
if sequence <= state.last_ingress_sequence:
@@ -1154,17 +1238,16 @@ def run(
if not token or len(token) < 40:
raise RuntimeError("LAB E15 shadow token is missing")
- def report_stage(stage: str, frame_index: int | None = None) -> None:
- if runtime_state is None:
- return
- runtime_state["current_stage"] = stage
- runtime_state["stage_observed_at_utc"] = (
- datetime.now(UTC).isoformat().replace("+00:00", "Z")
- )
- if frame_index is not None:
- runtime_state["active_frame_index"] = frame_index
+ stage_telemetry = (
+ runtime_state.get("_stage_telemetry")
+ if runtime_state is not None
+ else None
+ )
+ if not isinstance(stage_telemetry, _StageExecutionTelemetry):
+ stage_telemetry = _StageExecutionTelemetry()
+ if runtime_state is not None:
+ runtime_state["_stage_telemetry"] = stage_telemetry
- report_stage("preprocessing")
common = _common(args) if loaded is None else loaded.common
live = common["live"]
e14 = common["e14"]
@@ -1231,7 +1314,6 @@ def run(
def on_decoded(frame: DecodedCameraFrame) -> None:
nonlocal decoded_frame_count
try:
- report_stage("camera-decode", frame.frame_index)
if not first_camera_epoch_ns:
first_camera_epoch_ns.append(frame.metadata.captured_at_epoch_ns)
last_camera_epoch_ns[:] = [frame.metadata.captured_at_epoch_ns]
@@ -1269,6 +1351,10 @@ def run(
height=600,
maximum_buffer_bytes=int(live["transport"]["maximum_media_buffer_bytes"]),
metadata_capacity=int(live["transport"]["camera_metadata_capacity"]),
+ measure_decode=lambda frame_index: stage_telemetry.measure(
+ "camera-decode",
+ frame_index,
+ ),
)
transport = _TransportState(Counter(), Counter())
result_queue: queue.Queue[bytes] = queue.Queue(maxsize=2)
@@ -1378,6 +1464,11 @@ def run(
self.count += 1
completed_semantics = SemanticResultStream()
+
+ def monitored_semantic_inference(image: Any) -> Any:
+ with stage_telemetry.measure("semantic-model"):
+ return infer_semantic(image)
+
semantic_thread = threading.Thread(
target=semantic_worker,
kwargs={
@@ -1386,7 +1477,7 @@ def run(
"valid_mask": valid_mask,
"target_lut": target_lut,
"target_names": target_names,
- "infer": infer_semantic,
+ "infer": monitored_semantic_inference,
"latency": semantic_latency,
"completed": completed_semantics,
"failures": semantic_errors,
@@ -1400,7 +1491,6 @@ def run(
)
semantic_thread.start()
decoder.start()
- report_stage("source-ingress")
receiver_thread = threading.Thread(
target=_receiver,
kwargs={
@@ -1418,6 +1508,7 @@ def run(
"sensor_decode_ms": sensor_decode_ms,
"result_queue": result_queue,
"result_complete": result_complete,
+ "stage_telemetry": stage_telemetry,
},
name="lab-e15-shadow-receiver",
daemon=True,
@@ -1447,14 +1538,22 @@ def run(
latency["decode_age_ms"].append(float(envelope.decode_ms))
latency["queue_wait_ms"].append(max(0.0, (started - envelope.decoded_monotonic) * 1000))
try:
- report_stage("preprocessing", envelope.frame_index)
detector_started = time.perf_counter()
- tensor = _preprocess(envelope.image, valid_mask, detector)
- report_stage("detector", envelope.frame_index)
- output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor)
- detections, _rejected = _detections(output_tensor, detector, valid_mask)
- report_stage("tracking", envelope.frame_index)
- tracks = tracker.update(detections, envelope.frame_index)
+ with stage_telemetry.measure("preprocessing", envelope.frame_index):
+ tensor = _preprocess(envelope.image, valid_mask, detector)
+ with stage_telemetry.measure("detector", envelope.frame_index):
+ output_tensor, _request_ms = _infer(
+ args.triton_url,
+ detector["model"],
+ tensor,
+ )
+ detections, _rejected = _detections(
+ output_tensor,
+ detector,
+ valid_mask,
+ )
+ with stage_telemetry.measure("tracking", envelope.frame_index):
+ tracks = tracker.update(detections, envelope.frame_index)
latency["detector_ms"].append((time.perf_counter() - detector_started) * 1000)
frame_seconds = float(envelope.timeline["session_seconds"])
@@ -1484,35 +1583,35 @@ def run(
).reshape((-1, 3))
position = binding.pose.position_xyz
quaternion = binding.pose.orientation_xyzw
- report_stage("sensor-fusion", envelope.frame_index)
- projection_started = time.perf_counter()
- pixels, depths, source_indices, points_lidar = project_points(
- points_map,
- position,
- quaternion,
- projection,
- )
- latency["projection_ms"].append(
- (time.perf_counter() - projection_started) * 1000
- )
- association_started = time.perf_counter()
- fusions = fuse_tracks(
- tracks=[_track_document(track) for track in tracks],
- semantic_map=current_semantic.mask,
- pixels=pixels,
- depths=depths,
- source_indices=source_indices,
- points_map=points_map,
- points_lidar=points_lidar,
- association=e14["association"],
- distance_history=history,
- completion_tracker=completion_tracker,
- sensor_position_map=position,
- session_seconds=frame_seconds,
- )
- latency["association_ms"].append(
- (time.perf_counter() - association_started) * 1000
- )
+ with stage_telemetry.measure("sensor-fusion", envelope.frame_index):
+ projection_started = time.perf_counter()
+ pixels, depths, source_indices, points_lidar = project_points(
+ points_map,
+ position,
+ quaternion,
+ projection,
+ )
+ latency["projection_ms"].append(
+ (time.perf_counter() - projection_started) * 1000
+ )
+ association_started = time.perf_counter()
+ fusions = fuse_tracks(
+ tracks=[_track_document(track) for track in tracks],
+ semantic_map=current_semantic.mask,
+ pixels=pixels,
+ depths=depths,
+ source_indices=source_indices,
+ points_map=points_map,
+ points_lidar=points_lidar,
+ association=e14["association"],
+ distance_history=history,
+ completion_tracker=completion_tracker,
+ sensor_position_map=position,
+ session_seconds=frame_seconds,
+ )
+ latency["association_ms"].append(
+ (time.perf_counter() - association_started) * 1000
+ )
fusion_state = "fused"
fused_frames += 1
fusion_state_counts[fusion_state] += 1
@@ -1555,21 +1654,21 @@ def run(
if temporal_stabilizer is None:
fusion_objects = raw_fusion_objects
else:
- report_stage("temporal-state", envelope.frame_index)
- temporal_started = time.perf_counter()
- fusion_objects = temporal_stabilizer.update(
- frame_index=envelope.frame_index,
- session_seconds=frame_seconds,
- objects=raw_fusion_objects,
- )
- world = stabilize_world_state(
- world,
- fusion_objects,
- temporal_world_memory,
- )
- latency["temporal_2d_3d_ms"].append(
- (time.perf_counter() - temporal_started) * 1000
- )
+ with stage_telemetry.measure("temporal-state", envelope.frame_index):
+ temporal_started = time.perf_counter()
+ fusion_objects = temporal_stabilizer.update(
+ frame_index=envelope.frame_index,
+ session_seconds=frame_seconds,
+ objects=raw_fusion_objects,
+ )
+ world = stabilize_world_state(
+ world,
+ fusion_objects,
+ temporal_world_memory,
+ )
+ latency["temporal_2d_3d_ms"].append(
+ (time.perf_counter() - temporal_started) * 1000
+ )
stabilized_cuboids += sum(
str(item.get("cuboid_status", "")).startswith("accepted-")
for item in fusion_objects
@@ -1628,30 +1727,29 @@ def run(
quality=80,
optimize=False,
)
- live_result = encode_live_perception_result(
- frame_index=envelope.frame_index,
- source_frame_index=int(envelope.timeline["source_frame_index"]),
- session_seconds=frame_seconds,
- captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
- image_jpeg=encoded_image.getvalue(),
- segmentation_mask=(
- current_semantic.mask
- if current_semantic is not None and semantic_status == "fresh"
- else None
- ),
- objects=fusion_objects,
- delivery=world["delivery"],
- )
- report_stage("result-publication", envelope.frame_index)
- try:
- result_queue.put_nowait(live_result)
- except queue.Full:
- with suppress(queue.Empty):
- result_queue.get_nowait()
- result_queue.task_done()
- transport.results_dropped += 1
- result_queue.put_nowait(live_result)
- report_stage("source-ingress", envelope.frame_index)
+ with stage_telemetry.measure("result-publication", envelope.frame_index):
+ live_result = encode_live_perception_result(
+ frame_index=envelope.frame_index,
+ source_frame_index=int(envelope.timeline["source_frame_index"]),
+ session_seconds=frame_seconds,
+ captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
+ image_jpeg=encoded_image.getvalue(),
+ segmentation_mask=(
+ current_semantic.mask
+ if current_semantic is not None and semantic_status == "fresh"
+ else None
+ ),
+ objects=fusion_objects,
+ delivery=world["delivery"],
+ )
+ try:
+ result_queue.put_nowait(live_result)
+ except queue.Full:
+ with suppress(queue.Empty):
+ result_queue.get_nowait()
+ result_queue.task_done()
+ transport.results_dropped += 1
+ result_queue.put_nowait(live_result)
except Exception:
detector_failures += 1
raise
@@ -1907,6 +2005,7 @@ def run(
},
"gpu_telemetry": gpu.summary(),
"runtime_telemetry": runtime_summary,
+ "stage_telemetry": stage_telemetry.snapshot(),
"process_cpu_seconds": time.process_time() - process_cpu_started,
"process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024,
"cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20,
@@ -2094,8 +2193,7 @@ def serve(args: argparse.Namespace) -> int:
"failed_runs": 0,
"active_request_id": None,
"active_frame_index": None,
- "current_stage": None,
- "stage_observed_at_utc": None,
+ "_stage_telemetry": _StageExecutionTelemetry(),
}
class Handler(BaseHTTPRequestHandler):
@@ -2124,6 +2222,7 @@ def serve(args: argparse.Namespace) -> int:
if self.path != "/health":
self._send(404, {"ok": False, "error": "not-found"})
return
+ stage_snapshot = state["_stage_telemetry"].snapshot()
self._send(
200,
{
@@ -2134,9 +2233,10 @@ def serve(args: argparse.Namespace) -> int:
"completed_runs": state["completed_runs"],
"failed_runs": state["failed_runs"],
"active_request_id": state["active_request_id"],
- "active_frame_index": state["active_frame_index"],
- "current_stage": state["current_stage"],
- "stage_observed_at_utc": state["stage_observed_at_utc"],
+ "active_frame_index": stage_snapshot["active_frame_index"],
+ "current_stage": stage_snapshot["current_stage"],
+ "active_stages": stage_snapshot["active_stages"],
+ "stage_metrics": stage_snapshot["stages"],
"authority": common["live"]["authority"],
"gpu": torch.cuda.get_device_name(),
},
@@ -2167,6 +2267,7 @@ def serve(args: argparse.Namespace) -> int:
run_args = _persistent_run_arguments(args, document)
document["token"] = None
state["active_request_id"] = request_id
+ state["_stage_telemetry"] = _StageExecutionTelemetry()
exit_code = run(run_args, loaded, state)
state["completed_runs"] += 1
self._send(
@@ -2203,9 +2304,6 @@ def serve(args: argparse.Namespace) -> int:
finally:
state["busy"] = False
state["active_request_id"] = None
- state["active_frame_index"] = None
- state["current_stage"] = None
- state["stage_observed_at_utc"] = None
run_lock.release()
server = ThreadingHTTPServer((args.listen_host, args.listen_port), Handler)
diff --git a/src/k1link/web/system_telemetry_api.py b/src/k1link/web/system_telemetry_api.py
index 5fd4af4..d34c32c 100644
--- a/src/k1link/web/system_telemetry_api.py
+++ b/src/k1link/web/system_telemetry_api.py
@@ -499,6 +499,12 @@ def _pipeline_document(raw: dict[str, Any]) -> dict[str, Any]:
current_stage = perception.get("current_stage")
if not isinstance(current_stage, str):
current_stage = None
+ active_stages = {
+ value
+ for value in _items(perception.get("active_stages"))
+ if isinstance(value, str)
+ }
+ stage_metrics = _mapping(perception.get("stage_metrics"))
busy = perception.get("state") == "busy"
stages = (
("source-ingress", "Приём сенсорного потока"),
@@ -517,6 +523,8 @@ def _pipeline_document(raw: dict[str, Any]) -> dict[str, Any]:
return "unavailable"
if not busy:
return "ready"
+ if active_stages:
+ return "active" if stage_id in active_stages else "waiting"
if stage_id == current_stage or stage_id in {
"source-ingress",
"camera-decode",
@@ -525,6 +533,30 @@ def _pipeline_document(raw: dict[str, Any]) -> dict[str, Any]:
return "active"
return "waiting"
+ def stage_document(stage_id: str, label: str) -> dict[str, Any]:
+ raw_metric = _mapping(stage_metrics.get(stage_id))
+ elapsed_seconds = _number(raw_metric.get("elapsed_seconds"))
+ share_percent = _number(raw_metric.get("share_percent"))
+ activations = raw_metric.get("activations")
+ return {
+ "id": stage_id,
+ "label": label,
+ "state": stage_state(stage_id),
+ "elapsed_seconds": (
+ max(0.0, elapsed_seconds) if elapsed_seconds is not None else None
+ ),
+ "share_percent": (
+ min(100.0, max(0.0, share_percent))
+ if share_percent is not None
+ else None
+ ),
+ "activations": (
+ max(0, int(activations))
+ if isinstance(activations, int) and not isinstance(activations, bool)
+ else None
+ ),
+ }
+
return {
"service_state": (
perception.get("state") if isinstance(perception.get("state"), str) else "unavailable"
@@ -551,14 +583,7 @@ def _pipeline_document(raw: dict[str, Any]) -> dict[str, Any]:
else None
),
"model_load_seconds": _number(perception.get("model_load_seconds")),
- "stages": [
- {
- "id": stage_id,
- "label": label,
- "state": stage_state(stage_id),
- }
- for stage_id, label in stages
- ],
+ "stages": [stage_document(stage_id, label) for stage_id, label in stages],
}
diff --git a/tests/test_e15_shadow_inference.py b/tests/test_e15_shadow_inference.py
index b29815f..506ff0b 100644
--- a/tests/test_e15_shadow_inference.py
+++ b/tests/test_e15_shadow_inference.py
@@ -394,3 +394,31 @@ def test_runtime_telemetry_captures_bounded_queue_snapshots() -> None:
row = json.loads(stream.getvalue())
assert row["schema_version"] == "missioncore.worker-runtime-telemetry/v1"
assert row["queues"]["detector"]["maximum_depth"] == 1
+
+
+def test_stage_execution_telemetry_measures_named_spans_without_process_claims() -> None:
+ module = _module()
+ telemetry = module._StageExecutionTelemetry(("detector", "tracking"))
+
+ with telemetry.measure("detector", frame_index=42):
+ pass
+ with telemetry.measure("tracking", frame_index=42):
+ pass
+
+ snapshot = telemetry.snapshot()
+ assert snapshot["current_stage"] is None
+ assert snapshot["active_stages"] == []
+ assert snapshot["active_frame_index"] == 42
+ assert snapshot["stages"]["detector"]["activations"] == 1
+ assert snapshot["stages"]["tracking"]["activations"] == 1
+ assert snapshot["stages"]["detector"]["elapsed_seconds"] >= 0
+ assert snapshot["stages"]["tracking"]["elapsed_seconds"] >= 0
+ assert sum(
+ stage["share_percent"] for stage in snapshot["stages"].values()
+ ) == pytest.approx(100)
+
+ with (
+ pytest.raises(RuntimeError, match="unknown pipeline stage"),
+ telemetry.measure("unregistered"),
+ ):
+ pass
diff --git a/tests/test_system_telemetry_api.py b/tests/test_system_telemetry_api.py
index cf54ab1..0d9a080 100644
--- a/tests/test_system_telemetry_api.py
+++ b/tests/test_system_telemetry_api.py
@@ -121,11 +121,24 @@ def _probe(
"perception": {
"state": "busy",
"current_stage": "detector",
+ "active_stages": ["detector", "semantic-model"],
"active_request_id": "run-001",
"active_frame_index": 42,
"completed_runs": 3,
"failed_runs": 0,
"model_load_seconds": 10.5,
+ "stage_metrics": {
+ "detector": {
+ "elapsed_seconds": 2.5,
+ "activations": 42,
+ "share_percent": 62.5,
+ },
+ "semantic-model": {
+ "elapsed_seconds": 1.5,
+ "activations": 11,
+ "share_percent": 37.5,
+ },
+ },
},
},
}
@@ -203,16 +216,25 @@ def test_worker_telemetry_separates_mission_core_and_external_load(
assert runtimes["sentinel-frigate"]["external"] is True
assert runtimes["sentinel-frigate"]["cpu_percent"] == 150
assert document["pipeline"]["active_request_id"] == "run-001"
- assert next(
+ detector_stage = next(
stage
for stage in document["pipeline"]["stages"]
if stage["id"] == "detector"
- )["state"] == "active"
+ )
+ assert detector_stage["state"] == "active"
+ assert detector_stage["elapsed_seconds"] == 2.5
+ assert detector_stage["activations"] == 42
+ assert detector_stage["share_percent"] == 62.5
assert next(
stage
for stage in document["pipeline"]["stages"]
if stage["id"] == "semantic-model"
)["state"] == "active"
+ assert next(
+ stage
+ for stage in document["pipeline"]["stages"]
+ if stage["id"] == "preprocessing"
+ )["share_percent"] is None
def test_profile_apply_fails_closed_on_wrong_node_and_keeps_old_profile(