feat(lab): publish M4.8S fixed-class detector replay
This commit is contained in:
DCCONSTRUCTIONS
@@ -125,6 +125,9 @@ from k1link.web.lidar_api import build_lidar_router
|
||||
from k1link.web.lidar_local_surface_service import K1LocalSurfaceReadService
|
||||
from k1link.web.m4_threat_replay_api import build_m4_threat_replay_router
|
||||
from k1link.web.m48_object_quality_api import build_m48_object_quality_router
|
||||
from k1link.web.m48s_fixed_class_detector_lab_api import (
|
||||
build_m48s_fixed_class_detector_lab_router,
|
||||
)
|
||||
from k1link.web.map_api import (
|
||||
MapGatewayConfiguration,
|
||||
MapGatewayProxy,
|
||||
@@ -944,6 +947,23 @@ app.include_router(
|
||||
),
|
||||
)
|
||||
)
|
||||
app.include_router(
|
||||
build_m48s_fixed_class_detector_lab_router(
|
||||
root_provider=lambda: (
|
||||
REPOSITORY_ROOT
|
||||
/ ".runtime"
|
||||
/ "compute-experiments"
|
||||
/ "m48s-semantic-shadow"
|
||||
/ "fixed-class-detector-lab-results"
|
||||
),
|
||||
repository_root_provider=lambda: REPOSITORY_ROOT,
|
||||
camera_frame_provider=(
|
||||
session_recorded_camera_frame_service.extract
|
||||
if session_recorded_camera_frame_service is not None
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
app.include_router(
|
||||
build_e47_semantic_slam_router(
|
||||
root_provider=lambda: (
|
||||
|
||||
@@ -0,0 +1,523 @@
|
||||
"""Read-only API for the sealed M4.8S fixed-class detector LAB."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from functools import lru_cache
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any, Final
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, Response
|
||||
from fastapi.responses import FileResponse
|
||||
|
||||
from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
|
||||
from k1link.laboratory.evidence_report import (
|
||||
LaboratoryEvidenceReportError,
|
||||
verify_laboratory_evidence_result,
|
||||
)
|
||||
from k1link.laboratory.m48s_fixed_class_detector_lab import (
|
||||
CATALOG_SCHEMA,
|
||||
FRAME_SCHEMA,
|
||||
INTEGRATED_STATUS,
|
||||
LAB_SCHEMA,
|
||||
RESULT_PREFIX,
|
||||
)
|
||||
from k1link.perception.fixed_class_detector_tournament import false_authority
|
||||
from k1link.perception.m48s_replay_timeline import (
|
||||
M48sReplayTimeline,
|
||||
M48sReplayTimelineError,
|
||||
)
|
||||
from k1link.perception.threat_timeline import RECORDED_SPATIAL_MAX_CHUNK_FRAMES
|
||||
from k1link.sessions import RecordedCameraFrame, SessionIntegrityError
|
||||
|
||||
RootProvider = Callable[[], Path | None]
|
||||
CameraFrameProvider = Callable[[str, int], RecordedCameraFrame]
|
||||
|
||||
RESULT_ID: Final = re.compile(rf"^{re.escape(RESULT_PREFIX)}[a-f0-9]{{64}}$")
|
||||
FRAME_ID: Final = re.compile(r"^[0-9]{6}$")
|
||||
SHA256: Final = re.compile(r"^[a-f0-9]{64}$")
|
||||
RESULT_PROJECTION_SCHEMA: Final = "missioncore.m48s-fixed-class-detector-result-view/v1"
|
||||
RESULT_CATALOG_SCHEMA: Final = "missioncore.m48s-fixed-class-detector-result-catalog/v1"
|
||||
MAX_JSON_BYTES: Final = 16 * 1024 * 1024
|
||||
MAX_FRAMES: Final = 32
|
||||
_DEFINITION: Final = LaboratoryEvidenceDefinition(
|
||||
work_id="m48s-fixed-class-detector",
|
||||
runtime_relative_root=PurePosixPath("m48s-semantic-shadow/fixed-class-detector-lab-results"),
|
||||
result_id_prefix="m48s-fixed-class-detector-lab",
|
||||
document_name="manifest.json",
|
||||
result_schema_version=LAB_SCHEMA,
|
||||
)
|
||||
|
||||
|
||||
def build_m48s_fixed_class_detector_lab_router(
|
||||
*,
|
||||
root_provider: RootProvider = lambda: None,
|
||||
repository_root_provider: RootProvider = lambda: None,
|
||||
camera_frame_provider: CameraFrameProvider | None = None,
|
||||
) -> APIRouter:
|
||||
router = APIRouter(
|
||||
prefix="/api/v1/laboratory/m48s/fixed-class-detector",
|
||||
tags=["laboratory"],
|
||||
)
|
||||
|
||||
def timeline(result_id: str) -> M48sReplayTimeline:
|
||||
candidate = _resolve_candidate(root_provider, result_id)
|
||||
repository_root = _configured_root(repository_root_provider)
|
||||
if repository_root is None:
|
||||
raise HTTPException(status_code=503, detail="M4.8S timeline source unavailable")
|
||||
try:
|
||||
return _read_timeline_cached(
|
||||
str(repository_root),
|
||||
str(candidate),
|
||||
result_id,
|
||||
_timeline_signature(candidate),
|
||||
)
|
||||
except (OSError, ValueError, M48sReplayTimelineError):
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="M4.8S bounded replay timeline failed verification",
|
||||
) from None
|
||||
|
||||
@router.get("/results")
|
||||
def list_results(limit: int = Query(default=1, ge=1, le=10)) -> dict[str, object]:
|
||||
root = _configured_root(root_provider)
|
||||
if root is None:
|
||||
return _empty_catalog(False)
|
||||
candidates = _candidates(root)
|
||||
items: list[dict[str, object]] = []
|
||||
invalid_total = 0
|
||||
for candidate in candidates:
|
||||
try:
|
||||
items.append(_project_result(candidate))
|
||||
except RuntimeError:
|
||||
invalid_total += 1
|
||||
items.sort(
|
||||
key=lambda item: (str(item["created_at_utc"]), str(item["result_id"])),
|
||||
reverse=True,
|
||||
)
|
||||
return {
|
||||
"schema_version": RESULT_CATALOG_SCHEMA,
|
||||
"configured": True,
|
||||
"items": items[:limit],
|
||||
"candidate_total": len(candidates),
|
||||
"invalid_total": invalid_total,
|
||||
"access": "read-only",
|
||||
}
|
||||
|
||||
@router.get("/{result_id}")
|
||||
def get_result(result_id: str) -> dict[str, object]:
|
||||
return _project_result(_resolve_candidate(root_provider, result_id))
|
||||
|
||||
@router.get("/{result_id}/frames/{frame_id}")
|
||||
def get_frame(result_id: str, frame_id: str) -> dict[str, object]:
|
||||
candidate, descriptor = _resolve_frame(
|
||||
root_provider,
|
||||
result_id=result_id,
|
||||
frame_id=frame_id,
|
||||
)
|
||||
path = candidate / str(descriptor["detail_path"])
|
||||
payload = _read_object(path)
|
||||
if (
|
||||
payload.get("schema_version") != FRAME_SCHEMA
|
||||
or payload.get("result_id") != result_id
|
||||
or payload.get("frame_id") != frame_id
|
||||
or descriptor.get("detail_sha256") != _sha256(path)
|
||||
or descriptor.get("detail_byte_length") != path.stat().st_size
|
||||
or not _valid_frame_payload(payload)
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="M4.8S frame not found")
|
||||
return {**copy.deepcopy(payload), "access": "read-only"}
|
||||
|
||||
@router.get("/{result_id}/frames/{frame_id}/camera")
|
||||
def get_camera(result_id: str, frame_id: str) -> FileResponse:
|
||||
candidate, descriptor = _resolve_frame(
|
||||
root_provider,
|
||||
result_id=result_id,
|
||||
frame_id=frame_id,
|
||||
)
|
||||
path = (candidate / str(descriptor["camera_path"])).resolve()
|
||||
if (
|
||||
not path.is_relative_to(candidate)
|
||||
or path.is_symlink()
|
||||
or not path.is_file()
|
||||
or descriptor.get("camera_sha256") != _sha256(path)
|
||||
or descriptor.get("camera_byte_length") != path.stat().st_size
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="M4.8S camera not found")
|
||||
return FileResponse(path, media_type="image/jpeg")
|
||||
|
||||
@router.get("/{result_id}/timeline")
|
||||
def get_timeline(result_id: str) -> dict[str, object]:
|
||||
return copy.deepcopy(timeline(result_id).metadata())
|
||||
|
||||
@router.get("/{result_id}/timeline/chunk")
|
||||
def get_timeline_chunk(
|
||||
result_id: str,
|
||||
start: int = Query(default=0, ge=0),
|
||||
count: int = Query(default=12, ge=1, le=RECORDED_SPATIAL_MAX_CHUNK_FRAMES),
|
||||
) -> dict[str, object]:
|
||||
try:
|
||||
return timeline(result_id).chunk(start_sequence=start, frame_count=count)
|
||||
except M48sReplayTimelineError:
|
||||
raise HTTPException(status_code=404, detail="M4.8S timeline chunk not found") from None
|
||||
|
||||
@router.get("/{result_id}/timeline/frames/{sequence}/camera")
|
||||
def get_timeline_camera(result_id: str, sequence: int) -> Response:
|
||||
if camera_frame_provider is None:
|
||||
raise HTTPException(status_code=503, detail="M4.8S camera decoder unavailable")
|
||||
projected = timeline(result_id)
|
||||
if not 0 <= sequence < len(projected.source_times_ns):
|
||||
raise HTTPException(status_code=404, detail="M4.8S timeline frame not found")
|
||||
try:
|
||||
camera = camera_frame_provider(projected.profile.session_id, sequence)
|
||||
except (OSError, SessionIntegrityError, ValueError):
|
||||
raise HTTPException(status_code=503, detail="M4.8S exact camera unavailable") from None
|
||||
if camera.width != 800 or camera.height != 600:
|
||||
raise HTTPException(status_code=503, detail="M4.8S camera size contract changed")
|
||||
return Response(
|
||||
content=camera.payload,
|
||||
media_type=camera.media_type,
|
||||
headers={
|
||||
"Cache-Control": "private, max-age=31536000, immutable",
|
||||
"ETag": f'"{camera.sha256}"',
|
||||
"X-Content-Type-Options": "nosniff",
|
||||
},
|
||||
)
|
||||
|
||||
return router
|
||||
|
||||
|
||||
@lru_cache(maxsize=4)
|
||||
def _read_timeline_cached(
|
||||
repository_root: str,
|
||||
result_root: str,
|
||||
result_id: str,
|
||||
signature: tuple[int, ...],
|
||||
) -> M48sReplayTimeline:
|
||||
del signature
|
||||
return M48sReplayTimeline(
|
||||
repository_root=Path(repository_root),
|
||||
result_root=Path(result_root),
|
||||
result_id=result_id,
|
||||
)
|
||||
|
||||
|
||||
def _timeline_signature(candidate: Path) -> tuple[int, ...]:
|
||||
signature: list[int] = []
|
||||
for name in (
|
||||
"manifest.json",
|
||||
"reference-graph-replay-frames.jsonl",
|
||||
"reference-graph-replay-worker-result.json",
|
||||
):
|
||||
path = candidate / name
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise ValueError("M4.8S replay artifact is unavailable")
|
||||
stat = path.stat()
|
||||
signature.extend((stat.st_size, stat.st_mtime_ns))
|
||||
return tuple(signature)
|
||||
|
||||
|
||||
def _project_result(candidate: Path) -> dict[str, object]:
|
||||
loaded = _load_result(candidate)
|
||||
manifest = loaded["manifest"]
|
||||
catalog = loaded["catalog"]
|
||||
identity = manifest["identity"]
|
||||
return {
|
||||
"schema_version": RESULT_PROJECTION_SCHEMA,
|
||||
"result_id": manifest["result_id"],
|
||||
"created_at_utc": manifest["created_at_utc"],
|
||||
"status": manifest["status"],
|
||||
"bounded_question_accepted": manifest["bounded_question_accepted"],
|
||||
"ground_truth": manifest["ground_truth"],
|
||||
"source": copy.deepcopy(identity["source"]),
|
||||
"configuration": copy.deepcopy(identity["configuration"]),
|
||||
"method": copy.deepcopy(manifest["method"]),
|
||||
"metrics": copy.deepcopy(manifest["metrics"]),
|
||||
"decision": copy.deepcopy(manifest["decision"]),
|
||||
"limitations": copy.deepcopy(manifest["limitations"]),
|
||||
"authority": copy.deepcopy(manifest["authority"]),
|
||||
"frames": copy.deepcopy(catalog["frames"]),
|
||||
"access": "read-only",
|
||||
}
|
||||
|
||||
|
||||
def _load_result(candidate: Path) -> dict[str, Any]:
|
||||
try:
|
||||
signature = _candidate_signature(candidate)
|
||||
except (OSError, RuntimeError, ValueError) as exc:
|
||||
raise RuntimeError("M4.8S result signature failed") from exc
|
||||
return _load_result_cached(str(candidate), signature)
|
||||
|
||||
|
||||
@lru_cache(maxsize=8)
|
||||
def _load_result_cached(candidate_value: str, signature: tuple[int, ...]) -> dict[str, Any]:
|
||||
del signature
|
||||
return _load_result_uncached(Path(candidate_value))
|
||||
|
||||
|
||||
def _load_result_uncached(candidate: Path) -> dict[str, Any]:
|
||||
if (
|
||||
not candidate.is_dir()
|
||||
or candidate.is_symlink()
|
||||
or RESULT_ID.fullmatch(candidate.name) is None
|
||||
):
|
||||
raise RuntimeError("M4.8S result candidate is invalid")
|
||||
try:
|
||||
verify_laboratory_evidence_result(_DEFINITION, candidate)
|
||||
except LaboratoryEvidenceReportError as exc:
|
||||
raise RuntimeError("M4.8S result integrity failed") from exc
|
||||
manifest = _read_object(candidate / "manifest.json")
|
||||
identity = manifest.get("identity")
|
||||
catalog_descriptor = manifest.get("catalog")
|
||||
decision = manifest.get("decision")
|
||||
method = manifest.get("method")
|
||||
metrics = manifest.get("metrics")
|
||||
status = manifest.get("status")
|
||||
integrated = status == INTEGRATED_STATUS
|
||||
if (
|
||||
manifest.get("schema_version") != LAB_SCHEMA
|
||||
or manifest.get("result_id") != candidate.name
|
||||
or status
|
||||
not in {
|
||||
"detector-load-passed-reference-graph-shadow-only",
|
||||
INTEGRATED_STATUS,
|
||||
}
|
||||
or manifest.get("completed") is not True
|
||||
or manifest.get("bounded_question_accepted") is not True
|
||||
or manifest.get("ground_truth") is not False
|
||||
or not isinstance(manifest.get("created_at_utc"), str)
|
||||
or not isinstance(identity, dict)
|
||||
or identity.get("schema_version") != LAB_SCHEMA
|
||||
or manifest.get("identity_sha256") != hashlib.sha256(_canonical_json(identity)).hexdigest()
|
||||
or not candidate.name.endswith(str(manifest.get("identity_sha256")))
|
||||
or identity.get("authority") != false_authority()
|
||||
or manifest.get("authority") != false_authority()
|
||||
or not isinstance(decision, dict)
|
||||
or decision.get("bounded_question_accepted") is not True
|
||||
or decision.get("ready_for_reference_graph_shadow") is not True
|
||||
or decision.get("integrated_world_state_gate_evaluated") is not integrated
|
||||
or (integrated and decision.get("integrated_world_state_gate_passed") is not True)
|
||||
or (integrated and decision.get("detector_replacement_authorized") is not False)
|
||||
or decision.get("production_accepted") is not False
|
||||
or not isinstance(method, dict)
|
||||
or method.get("schema_version") != "missioncore.laboratory-method/v1"
|
||||
or method.get("completeness") != "complete"
|
||||
or not isinstance(metrics, dict)
|
||||
or (integrated and not isinstance(metrics.get("integrated_world_state"), dict))
|
||||
or not isinstance(manifest.get("limitations"), list)
|
||||
or not isinstance(catalog_descriptor, dict)
|
||||
or catalog_descriptor.get("path") != "catalog.json"
|
||||
):
|
||||
raise RuntimeError("M4.8S manifest is invalid")
|
||||
catalog_path = candidate / "catalog.json"
|
||||
if (
|
||||
catalog_descriptor.get("sha256") != _sha256(catalog_path)
|
||||
or catalog_descriptor.get("byte_length") != catalog_path.stat().st_size
|
||||
):
|
||||
raise RuntimeError("M4.8S catalog changed")
|
||||
catalog = _read_object(catalog_path)
|
||||
frames = catalog.get("frames")
|
||||
if (
|
||||
catalog.get("schema_version") != CATALOG_SCHEMA
|
||||
or catalog.get("result_id") != candidate.name
|
||||
or not isinstance(frames, list)
|
||||
or not 1 <= len(frames) <= MAX_FRAMES
|
||||
or catalog.get("frame_count") != len(frames)
|
||||
or len({item.get("frame_id") for item in frames if isinstance(item, dict)}) != len(frames)
|
||||
or any(not _valid_descriptor(item) for item in frames)
|
||||
):
|
||||
raise RuntimeError("M4.8S catalog is invalid")
|
||||
return {"manifest": manifest, "catalog": catalog}
|
||||
|
||||
|
||||
def _candidate_signature(candidate: Path) -> tuple[int, ...]:
|
||||
if not candidate.is_dir() or candidate.is_symlink():
|
||||
raise RuntimeError("M4.8S result candidate is invalid")
|
||||
manifest_path = candidate / "manifest.json"
|
||||
manifest = _read_object(manifest_path)
|
||||
artifacts = manifest.get("artifacts")
|
||||
if not isinstance(artifacts, list):
|
||||
raise RuntimeError("M4.8S artifact manifest is invalid")
|
||||
signature = [manifest_path.stat().st_size, manifest_path.stat().st_mtime_ns]
|
||||
for descriptor in artifacts:
|
||||
if not isinstance(descriptor, dict) or not isinstance(descriptor.get("path"), str):
|
||||
raise RuntimeError("M4.8S artifact descriptor is invalid")
|
||||
path = (candidate / descriptor["path"]).resolve(strict=True)
|
||||
if not path.is_relative_to(candidate) or path.is_symlink() or not path.is_file():
|
||||
raise RuntimeError("M4.8S artifact path is invalid")
|
||||
stat = path.stat()
|
||||
signature.extend((stat.st_size, stat.st_mtime_ns))
|
||||
return tuple(signature)
|
||||
|
||||
|
||||
def _resolve_candidate(root_provider: RootProvider, result_id: str) -> Path:
|
||||
if RESULT_ID.fullmatch(result_id) is None:
|
||||
raise HTTPException(status_code=404, detail="M4.8S result not found")
|
||||
root = _configured_root(root_provider)
|
||||
if root is None:
|
||||
raise HTTPException(status_code=404, detail="M4.8S result not found")
|
||||
candidate = root / result_id
|
||||
try:
|
||||
_load_result(candidate)
|
||||
except RuntimeError:
|
||||
raise HTTPException(status_code=404, detail="M4.8S result not found") from None
|
||||
return candidate
|
||||
|
||||
|
||||
def _resolve_frame(
|
||||
root_provider: RootProvider,
|
||||
*,
|
||||
result_id: str,
|
||||
frame_id: str,
|
||||
) -> tuple[Path, dict[str, Any]]:
|
||||
if FRAME_ID.fullmatch(frame_id) is None:
|
||||
raise HTTPException(status_code=404, detail="M4.8S frame not found")
|
||||
candidate = _resolve_candidate(root_provider, result_id)
|
||||
loaded = _load_result(candidate)
|
||||
try:
|
||||
descriptor = next(
|
||||
item for item in loaded["catalog"]["frames"] if item["frame_id"] == frame_id
|
||||
)
|
||||
except StopIteration:
|
||||
raise HTTPException(status_code=404, detail="M4.8S frame not found") from None
|
||||
return candidate, descriptor
|
||||
|
||||
|
||||
def _valid_descriptor(value: object) -> bool:
|
||||
if not isinstance(value, dict):
|
||||
return False
|
||||
frame_id = value.get("frame_id")
|
||||
counts = value.get("counts")
|
||||
return (
|
||||
isinstance(frame_id, str)
|
||||
and FRAME_ID.fullmatch(frame_id) is not None
|
||||
and value.get("source_sequence") == int(frame_id)
|
||||
and value.get("camera_path") == f"frames/frame-{frame_id}.jpg"
|
||||
and isinstance(value.get("camera_sha256"), str)
|
||||
and SHA256.fullmatch(value["camera_sha256"]) is not None
|
||||
and isinstance(value.get("camera_byte_length"), int)
|
||||
and value["camera_byte_length"] > 0
|
||||
and value.get("detail_path") == f"frame-{frame_id}.json"
|
||||
and isinstance(value.get("detail_sha256"), str)
|
||||
and SHA256.fullmatch(value["detail_sha256"]) is not None
|
||||
and isinstance(value.get("detail_byte_length"), int)
|
||||
and 0 < value["detail_byte_length"] <= MAX_JSON_BYTES
|
||||
and isinstance(counts, dict)
|
||||
and set(counts) == {"yolox", "dfine", "rf-detr"}
|
||||
and all(isinstance(count, int) and count >= 0 for count in counts.values())
|
||||
)
|
||||
|
||||
|
||||
def _valid_frame_payload(value: dict[str, Any]) -> bool:
|
||||
camera = value.get("camera")
|
||||
detections = value.get("detections")
|
||||
return (
|
||||
isinstance(value.get("source_sequence"), int)
|
||||
and isinstance(camera, dict)
|
||||
and camera.get("media_type") == "image/jpeg"
|
||||
and camera.get("width") == 800
|
||||
and camera.get("height") == 600
|
||||
and camera.get("exact_source_frame") is True
|
||||
and isinstance(camera.get("sha256"), str)
|
||||
and SHA256.fullmatch(camera["sha256"]) is not None
|
||||
and value.get("comparison_threshold") == 0.5
|
||||
and isinstance(detections, dict)
|
||||
and set(detections) == {"yolox", "dfine", "rf-detr"}
|
||||
and all(
|
||||
isinstance(items, list) and all(_valid_detection(item) for item in items)
|
||||
for items in detections.values()
|
||||
)
|
||||
and value.get("ground_truth_available") is False
|
||||
and value.get("authority") == false_authority()
|
||||
)
|
||||
|
||||
|
||||
def _valid_detection(value: object) -> bool:
|
||||
if not isinstance(value, dict) or set(value) != {"label", "score", "bbox_xyxy"}:
|
||||
return False
|
||||
score = value.get("score")
|
||||
bbox = value.get("bbox_xyxy")
|
||||
return (
|
||||
isinstance(value.get("label"), str)
|
||||
and isinstance(score, (int, float))
|
||||
and not isinstance(score, bool)
|
||||
and 0.5 <= float(score) <= 1.0
|
||||
and isinstance(bbox, list)
|
||||
and len(bbox) == 4
|
||||
and all(isinstance(item, (int, float)) and not isinstance(item, bool) for item in bbox)
|
||||
and 0 <= float(bbox[0]) < float(bbox[2]) <= 800
|
||||
and 0 <= float(bbox[1]) < float(bbox[3]) <= 600
|
||||
)
|
||||
|
||||
|
||||
def _configured_root(provider: RootProvider) -> Path | None:
|
||||
value = provider()
|
||||
if value is None:
|
||||
return None
|
||||
candidate = value.expanduser().absolute()
|
||||
if candidate.is_symlink():
|
||||
return None
|
||||
try:
|
||||
resolved = candidate.resolve(strict=True)
|
||||
except OSError:
|
||||
return None
|
||||
return resolved if resolved.is_dir() else None
|
||||
|
||||
|
||||
def _candidates(root: Path) -> list[Path]:
|
||||
return [
|
||||
item
|
||||
for item in root.iterdir()
|
||||
if item.is_dir() and not item.is_symlink() and RESULT_ID.fullmatch(item.name)
|
||||
]
|
||||
|
||||
|
||||
def _empty_catalog(configured: bool) -> dict[str, object]:
|
||||
return {
|
||||
"schema_version": RESULT_CATALOG_SCHEMA,
|
||||
"configured": configured,
|
||||
"items": [],
|
||||
"candidate_total": 0,
|
||||
"invalid_total": 0,
|
||||
"access": "read-only",
|
||||
}
|
||||
|
||||
|
||||
def _read_object(path: Path) -> dict[str, Any]:
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_JSON_BYTES:
|
||||
raise RuntimeError("M4.8S JSON artifact is invalid")
|
||||
try:
|
||||
value = json.loads(path.read_text("utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise RuntimeError("M4.8S JSON artifact is invalid") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise RuntimeError("M4.8S JSON artifact is invalid")
|
||||
return value
|
||||
|
||||
|
||||
def _canonical_json(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"RESULT_CATALOG_SCHEMA",
|
||||
"RESULT_PROJECTION_SCHEMA",
|
||||
"build_m48s_fixed_class_detector_lab_router",
|
||||
]
|
||||
Reference in New Issue
Block a user