SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: prove and decode K1 realtime MQTT streams

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-15 19:10:41 +03:00
parent faa442eefc
commit 6b22e5a1d2
30 changed files with 4375 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/analyze/__init__.py
+15
View File
@@ -0,0 +1,15 @@
"""Bounded, offline analysis of sensitive K1 evidence artifacts."""
from k1link.analyze.stream_summary import (
DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES,
MAX_STREAM_SUMMARY_PAYLOAD_BYTES,
StreamSummary,
summarize_mqtt_streams,
)
__all__ = [
"DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES",
"MAX_STREAM_SUMMARY_PAYLOAD_BYTES",
"StreamSummary",
"summarize_mqtt_streams",
]
src/k1link/analyze/stream_summary.py
+354
View File
@@ -0,0 +1,354 @@
from __future__ import annotations
import hashlib
import math
import os
import stat
from dataclasses import dataclass, field
from pathlib import Path
from typing import TypedDict
from k1link.artifacts import utc_now_iso
from k1link.mqtt import DEFAULT_MAX_MESSAGE_BYTES, iter_capture_frames
from k1link.protocol import (
DecodeLimits,
StreamDecodeError,
decode_lio_pcl,
decode_lio_pose,
)
LIO_PCL_TOPIC = "lixel/application/report/lio_pcl"
LIO_POSE_TOPIC = "lixel/application/report/lio_pose"
DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES = DecodeLimits().max_mqtt_payload_bytes
MAX_STREAM_SUMMARY_PAYLOAD_BYTES = DEFAULT_MAX_MESSAGE_BYTES
_HASH_CHUNK_BYTES = 1024 * 1024
class IntRangeSummary(TypedDict):
min: int | None
max: int | None
class ScalerSummary(TypedDict):
min: int | None
max: int | None
constant: bool | None
class ByteSummary(TypedDict):
total: int
per_frame: IntRangeSummary
class PointSummary(TypedDict):
total: int
per_frame: IntRangeSummary
class SourceSummary(TypedDict):
bytes: int
sha256: str
class LimitSummary(TypedDict):
max_payload_bytes: int
max_compressed_bytes: int
max_decompressed_bytes: int
max_compression_ratio: int
max_points_per_frame: int
class FrameSummary(TypedDict):
count: int
payload_bytes: int
encoded_frame_bytes: int
other_count: int
other_payload_bytes: int
class DecodeSummary(TypedDict):
attempted: int
successes: int
errors: int
class PointCloudSummary(TypedDict):
frame_count: int
payload_bytes: int
decode_successes: int
decode_errors: int
points: PointSummary
scalers: ScalerSummary
compressed_bytes: ByteSummary
decompressed_bytes: ByteSummary
class PoseSummary(TypedDict):
frame_count: int
payload_bytes: int
decode_successes: int
decode_errors: int
first_to_last_displacement_meters: float | None
class StreamSummary(TypedDict):
schema_version: int
created_at_utc: str
sensitivity: str
source: SourceSummary
limits: LimitSummary
frames: FrameSummary
decoding: DecodeSummary
point_cloud: PointCloudSummary
pose: PoseSummary
@dataclass(slots=True)
class _IntRange:
minimum: int | None = None
maximum: int | None = None
def add(self, value: int) -> None:
if self.minimum is None or value < self.minimum:
self.minimum = value
if self.maximum is None or value > self.maximum:
self.maximum = value
def summary(self) -> IntRangeSummary:
return {"min": self.minimum, "max": self.maximum}
@dataclass(slots=True)
class _PointCloudAccumulator:
frame_count: int = 0
payload_bytes: int = 0
decode_successes: int = 0
decode_errors: int = 0
point_total: int = 0
compressed_total: int = 0
decompressed_total: int = 0
points_per_frame: _IntRange = field(default_factory=_IntRange)
scalers: _IntRange = field(default_factory=_IntRange)
compressed_per_frame: _IntRange = field(default_factory=_IntRange)
decompressed_per_frame: _IntRange = field(default_factory=_IntRange)
first_scaler: int | None = None
scaler_constant: bool = True
def record_payload(self, payload_bytes: int) -> None:
self.frame_count += 1
self.payload_bytes += payload_bytes
def record_error(self) -> None:
self.decode_errors += 1
def record_decoded(
self,
*,
point_count: int,
scaler: int,
compressed_bytes: int,
decompressed_bytes: int,
) -> None:
self.decode_successes += 1
self.point_total += point_count
self.compressed_total += compressed_bytes
self.decompressed_total += decompressed_bytes
self.points_per_frame.add(point_count)
self.scalers.add(scaler)
self.compressed_per_frame.add(compressed_bytes)
self.decompressed_per_frame.add(decompressed_bytes)
if self.first_scaler is None:
self.first_scaler = scaler
elif scaler != self.first_scaler:
self.scaler_constant = False
def summary(self) -> PointCloudSummary:
scaler_summary: ScalerSummary = {
"min": self.scalers.minimum,
"max": self.scalers.maximum,
"constant": self.scaler_constant if self.decode_successes else None,
}
return {
"frame_count": self.frame_count,
"payload_bytes": self.payload_bytes,
"decode_successes": self.decode_successes,
"decode_errors": self.decode_errors,
"points": {
"total": self.point_total,
"per_frame": self.points_per_frame.summary(),
},
"scalers": scaler_summary,
"compressed_bytes": {
"total": self.compressed_total,
"per_frame": self.compressed_per_frame.summary(),
},
"decompressed_bytes": {
"total": self.decompressed_total,
"per_frame": self.decompressed_per_frame.summary(),
},
}
@dataclass(slots=True)
class _PoseAccumulator:
frame_count: int = 0
payload_bytes: int = 0
decode_successes: int = 0
decode_errors: int = 0
first_position: tuple[float, float, float] | None = None
last_position: tuple[float, float, float] | None = None
def record_payload(self, payload_bytes: int) -> None:
self.frame_count += 1
self.payload_bytes += payload_bytes
def record_error(self) -> None:
self.decode_errors += 1
def record_decoded(self, position: tuple[float, float, float]) -> None:
self.decode_successes += 1
if self.first_position is None:
self.first_position = position
self.last_position = position
def summary(self) -> PoseSummary:
displacement: float | None = None
if self.first_position is not None and self.last_position is not None:
candidate = math.dist(self.first_position, self.last_position)
if math.isfinite(candidate):
displacement = candidate
return {
"frame_count": self.frame_count,
"payload_bytes": self.payload_bytes,
"decode_successes": self.decode_successes,
"decode_errors": self.decode_errors,
"first_to_last_displacement_meters": displacement,
}
def summarize_mqtt_streams(
capture: Path,
*,
max_payload_bytes: int = DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES,
) -> StreamSummary:
"""Stream a raw MQTT capture into an aggregate-only, coordinate-free summary."""
if not 1 <= max_payload_bytes <= MAX_STREAM_SUMMARY_PAYLOAD_BYTES:
raise ValueError(
"max_payload_bytes must be between 1 and "
f"{MAX_STREAM_SUMMARY_PAYLOAD_BYTES}"
)
capture_path = capture.expanduser()
before = capture_path.stat()
if not stat.S_ISREG(before.st_mode):
raise ValueError("capture must be a regular file")
capture_sha256 = _sha256_file(capture_path)
hashed = capture_path.stat()
if _file_identity(before) != _file_identity(hashed):
raise RuntimeError("capture changed while it was being hashed")
limits = DecodeLimits(max_mqtt_payload_bytes=max_payload_bytes)
point_cloud = _PointCloudAccumulator()
pose = _PoseAccumulator()
frame_count = 0
payload_bytes = 0
encoded_frame_bytes = 0
other_count = 0
other_payload_bytes = 0
for capture_frame in iter_capture_frames(
capture_path,
max_payload_bytes=max_payload_bytes,
):
frame_count += 1
frame_payload_bytes = len(capture_frame.payload)
payload_bytes += frame_payload_bytes
encoded_frame_bytes += capture_frame.raw_frame_bytes
if capture_frame.topic == LIO_PCL_TOPIC:
point_cloud.record_payload(frame_payload_bytes)
try:
decoded = decode_lio_pcl(capture_frame.payload, limits)
except StreamDecodeError:
point_cloud.record_error()
continue
point_cloud.record_decoded(
point_count=len(decoded.points),
scaler=decoded.header.scaler,
compressed_bytes=decoded.compressed_bytes,
decompressed_bytes=decoded.decompressed_bytes,
)
continue
if capture_frame.topic == LIO_POSE_TOPIC:
pose.record_payload(frame_payload_bytes)
try:
decoded_pose = decode_lio_pose(capture_frame.payload, limits)
except StreamDecodeError:
pose.record_error()
continue
pose.record_decoded(decoded_pose.position_xyz)
continue
# Unknown topic text is intentionally neither retained nor emitted: a malformed
# capture could place an identifier in that field.
other_count += 1
other_payload_bytes += frame_payload_bytes
after = capture_path.stat()
if _file_identity(hashed) != _file_identity(after):
raise RuntimeError("capture changed while it was being analyzed")
decode_successes = point_cloud.decode_successes + pose.decode_successes
decode_errors = point_cloud.decode_errors + pose.decode_errors
return {
"schema_version": 1,
"created_at_utc": utc_now_iso(),
"sensitivity": (
"sensitive derived K1 stream statistics; keep in ignored storage; "
"identifiers, keys, error text and coordinates are omitted"
),
"source": {
"bytes": hashed.st_size,
"sha256": capture_sha256,
},
"limits": {
"max_payload_bytes": limits.max_mqtt_payload_bytes,
"max_compressed_bytes": limits.max_compressed_bytes,
"max_decompressed_bytes": limits.max_decompressed_bytes,
"max_compression_ratio": limits.max_compression_ratio,
"max_points_per_frame": limits.max_points_per_frame,
},
"frames": {
"count": frame_count,
"payload_bytes": payload_bytes,
"encoded_frame_bytes": encoded_frame_bytes,
"other_count": other_count,
"other_payload_bytes": other_payload_bytes,
},
"decoding": {
"attempted": point_cloud.frame_count + pose.frame_count,
"successes": decode_successes,
"errors": decode_errors,
},
"point_cloud": point_cloud.summary(),
"pose": pose.summary(),
}
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(_HASH_CHUNK_BYTES), b""):
digest.update(chunk)
return digest.hexdigest()
def _file_identity(file_stat: os.stat_result) -> tuple[int, int, int, int]:
return (
file_stat.st_dev,
file_stat.st_ino,
file_stat.st_size,
file_stat.st_mtime_ns,
)