feat(observatory): add laboratory setup preflight
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,19 @@
|
||||
"""Observation-only laboratory orchestration contracts."""
|
||||
|
||||
from k1link.observatory.canonical_result import (
|
||||
is_admitted_observatory_recorded_result,
|
||||
)
|
||||
from k1link.observatory.setups import (
|
||||
LABORATORY_SETUP_CATALOG_SCHEMA,
|
||||
LABORATORY_SETUP_REGISTRY_SCHEMA,
|
||||
LaboratorySetupRegistry,
|
||||
LaboratorySetupRegistryError,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"LABORATORY_SETUP_CATALOG_SCHEMA",
|
||||
"LABORATORY_SETUP_REGISTRY_SCHEMA",
|
||||
"LaboratorySetupRegistry",
|
||||
"LaboratorySetupRegistryError",
|
||||
"is_admitted_observatory_recorded_result",
|
||||
]
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Exact admission for the current canonical recorded Observatory result."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Final
|
||||
|
||||
from k1link.sessions.models import SessionSummary
|
||||
|
||||
_SHA256: Final = re.compile(r"^[a-f0-9]{64}$")
|
||||
_CANONICAL_RESULT: Final = re.compile(r"^lab-v1-vegetation-shadow-([a-f0-9]{64})$")
|
||||
_CANONICAL_SOURCE_SESSION_ID: Final = "20260828T130511Z_viewer_live"
|
||||
_CANONICAL_PIPELINE_ID: Final = (
|
||||
"ravnoves004tree-full-eomt-ddrnet-recorded-review/v1"
|
||||
)
|
||||
|
||||
|
||||
def is_admitted_observatory_recorded_result(
|
||||
summary: SessionSummary,
|
||||
*,
|
||||
expected_result_id: str,
|
||||
expected_source_session_id: str,
|
||||
expected_result_kind: str,
|
||||
) -> bool:
|
||||
"""Return true only for the exact capability-owned canonical projection."""
|
||||
|
||||
lab = summary.lab
|
||||
match = _CANONICAL_RESULT.fullmatch(expected_result_id)
|
||||
if (
|
||||
lab is None
|
||||
or match is None
|
||||
or expected_source_session_id != _CANONICAL_SOURCE_SESSION_ID
|
||||
or summary.session_id != expected_result_id
|
||||
or summary.origin != "missioncore.lab-instance/v1"
|
||||
or summary.status != "ready"
|
||||
or not summary.replayable
|
||||
or lab.session_id != expected_result_id
|
||||
or lab.result_id != expected_result_id
|
||||
or lab.source_session_id != expected_source_session_id
|
||||
or lab.lab_id != "LAB V1"
|
||||
or lab.result_kind != expected_result_kind
|
||||
or lab.result_kind != "recorded-perception-qualification"
|
||||
or lab.config_sha256 is not None
|
||||
or not isinstance(lab.source_result_id, str)
|
||||
or _CANONICAL_RESULT.fullmatch(lab.source_result_id) is None
|
||||
or lab.replay_capability is None
|
||||
):
|
||||
return False
|
||||
|
||||
capability = lab.replay_capability.as_dict()
|
||||
provenance = lab.provenance
|
||||
evidence_identity = match.group(1)
|
||||
if set(provenance) != {
|
||||
"schema_version",
|
||||
"evidence_identity_sha256",
|
||||
"result_document_sha256",
|
||||
"replay_capability",
|
||||
"authority",
|
||||
"method",
|
||||
}:
|
||||
return False
|
||||
result_document_sha256 = provenance.get("result_document_sha256")
|
||||
if (
|
||||
provenance.get("schema_version")
|
||||
!= "missioncore.canonical-recorded-lab-projection/v1"
|
||||
or provenance.get("evidence_identity_sha256") != evidence_identity
|
||||
or not isinstance(result_document_sha256, str)
|
||||
or _SHA256.fullmatch(result_document_sha256) is None
|
||||
or provenance.get("replay_capability") != capability
|
||||
or provenance.get("authority")
|
||||
!= {
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
"actuation_accepted": False,
|
||||
}
|
||||
):
|
||||
return False
|
||||
|
||||
return provenance.get("method") == {
|
||||
"schema_version": "missioncore.laboratory-method/v1",
|
||||
"completeness": "legacy-partial",
|
||||
"execution_class": "ai-inference",
|
||||
"pipeline_id": _CANONICAL_PIPELINE_ID,
|
||||
"components": [
|
||||
{
|
||||
"kind": "source",
|
||||
"name": "sealed full-route LAB result",
|
||||
"version": "missioncore.lab-v1-vegetation-shadow/v1",
|
||||
"role": "immutable Session catalog projection",
|
||||
"identity_sha256": evidence_identity,
|
||||
}
|
||||
],
|
||||
}
|
||||
@@ -0,0 +1,524 @@
|
||||
"""Strict setup catalog for the Observatory laboratory configurator.
|
||||
|
||||
The catalog can represent both a real historical RunDefinition and an exact
|
||||
sealed result that predates RunDefinitions. It never upgrades the latter into
|
||||
a fabricated executable configuration.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any, Final, Literal
|
||||
|
||||
from k1link.sessions.models import SessionSummary
|
||||
|
||||
LABORATORY_SETUP_REGISTRY_SCHEMA: Final = (
|
||||
"missioncore.observatory-laboratory-setup-registry/v1"
|
||||
)
|
||||
LABORATORY_SETUP_CATALOG_SCHEMA: Final = (
|
||||
"missioncore.observatory-laboratory-setup-catalog/v1"
|
||||
)
|
||||
_MAX_REGISTRY_BYTES: Final = 256 * 1024
|
||||
_MAX_CONFIGURATION_BYTES: Final = 4 * 1024 * 1024
|
||||
_IDENTIFIER: Final = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
|
||||
_RESULT_ID: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{2,159}$")
|
||||
_SHA256: Final = re.compile(r"^[a-f0-9]{64}$")
|
||||
_MODALITIES: Final = frozenset({"point-cloud", "trajectory", "video"})
|
||||
_AUTHORITY: Final = {
|
||||
"commands_enabled": False,
|
||||
"actuation_allowed": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
"production_accepted": False,
|
||||
}
|
||||
|
||||
SetupOrigin = Literal["archived-definition", "existing-result"]
|
||||
ResultAccess = Literal["legacy-lab", "observatory", "evidence-only"]
|
||||
|
||||
|
||||
class LaboratorySetupRegistryError(ValueError):
|
||||
"""A setup registry or its immutable references are invalid."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _ConfigurationReference:
|
||||
role: str
|
||||
path: PurePosixPath
|
||||
sha256: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _RunDefinition:
|
||||
definition_id: str
|
||||
version: int
|
||||
work_id: str
|
||||
configuration: tuple[_ConfigurationReference, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Executor:
|
||||
contour_id: str
|
||||
state: Literal["not-installed"]
|
||||
reason_code: str
|
||||
reason: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _PreservedResult:
|
||||
result_id: str
|
||||
result_kind: str
|
||||
relation: str
|
||||
access: ResultAccess
|
||||
created_at_utc: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LaboratorySetup:
|
||||
setup_id: str
|
||||
display_name: str
|
||||
description: str
|
||||
origin: SetupOrigin
|
||||
source_session_id: str
|
||||
source_label: str
|
||||
required_modalities: tuple[str, ...]
|
||||
run_definition: _RunDefinition | None
|
||||
executor: _Executor
|
||||
preserved_results: tuple[_PreservedResult, ...]
|
||||
|
||||
def project(
|
||||
self,
|
||||
source: SessionSummary,
|
||||
*,
|
||||
available_observatory_result_ids: frozenset[str],
|
||||
) -> dict[str, object]:
|
||||
reasons: list[dict[str, str]] = []
|
||||
if source.lab is not None:
|
||||
reasons.append(_reason("source-is-lab", "Нужна исходная, а не LAB-сессия."))
|
||||
if source.session_id != self.source_session_id:
|
||||
reasons.append(
|
||||
_reason(
|
||||
"source-session-not-admitted",
|
||||
"Эта версия привязана к другой запечатанной исходной сессии.",
|
||||
)
|
||||
)
|
||||
if source.display_name != self.source_label:
|
||||
reasons.append(
|
||||
_reason(
|
||||
"source-label-mismatch",
|
||||
"Идентичность источника не совпадает с сохранённым сетапом.",
|
||||
)
|
||||
)
|
||||
if source.status != "ready":
|
||||
reasons.append(
|
||||
_reason(
|
||||
"source-not-ready",
|
||||
"Исходная сессия не находится в запечатанном состоянии ready.",
|
||||
)
|
||||
)
|
||||
if not source.replayable:
|
||||
reasons.append(
|
||||
_reason(
|
||||
"source-not-replayable",
|
||||
"Для исходной сессии недоступно воспроизводимое чтение.",
|
||||
)
|
||||
)
|
||||
missing = sorted(set(self.required_modalities) - set(source.modalities))
|
||||
if missing:
|
||||
reasons.append(
|
||||
_reason(
|
||||
"required-modalities-unavailable",
|
||||
"Не хватает каналов: " + ", ".join(missing) + ".",
|
||||
)
|
||||
)
|
||||
compatible = not reasons
|
||||
ready_result_ids = [
|
||||
result.result_id
|
||||
for result in self.preserved_results
|
||||
if result.access == "observatory"
|
||||
and result.result_id in available_observatory_result_ids
|
||||
]
|
||||
if compatible and ready_result_ids:
|
||||
action = "open-existing"
|
||||
action_reason = "Точный запечатанный результат уже опубликован в Обсерватории."
|
||||
elif compatible and any(
|
||||
result.access == "legacy-lab" for result in self.preserved_results
|
||||
):
|
||||
action = "open-legacy"
|
||||
action_reason = "Точный результат сохранён в legacy LAB."
|
||||
else:
|
||||
action = "blocked"
|
||||
action_reason = reasons[0]["message"] if reasons else self.executor.reason
|
||||
return {
|
||||
"setup_id": self.setup_id,
|
||||
"display_name": self.display_name,
|
||||
"description": self.description,
|
||||
"origin": self.origin,
|
||||
"source": {
|
||||
"session_id": self.source_session_id,
|
||||
"label": self.source_label,
|
||||
"required_modalities": list(self.required_modalities),
|
||||
},
|
||||
"run_definition": _project_definition(self, self.run_definition),
|
||||
"compatibility": {"compatible": compatible, "reasons": reasons},
|
||||
"executor": {
|
||||
"contour_id": self.executor.contour_id,
|
||||
"state": self.executor.state,
|
||||
"reason_code": self.executor.reason_code,
|
||||
"reason": self.executor.reason,
|
||||
},
|
||||
"preserved_results": [
|
||||
{
|
||||
"result_id": result.result_id,
|
||||
"result_kind": result.result_kind,
|
||||
"relation": result.relation,
|
||||
"access": result.access,
|
||||
"created_at_utc": result.created_at_utc,
|
||||
"observatory_projection_available": (
|
||||
result.result_id in available_observatory_result_ids
|
||||
),
|
||||
}
|
||||
for result in self.preserved_results
|
||||
],
|
||||
"preflight": {
|
||||
"outcome": "existing" if action == "open-existing" else "blocked",
|
||||
"action": action,
|
||||
"reason": action_reason,
|
||||
"submission_allowed": False,
|
||||
"existing_result_ids": ready_result_ids,
|
||||
},
|
||||
"authority": dict(_AUTHORITY),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LaboratorySetupRegistry:
|
||||
setups: tuple[LaboratorySetup, ...]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
setup_ids = [setup.setup_id for setup in self.setups]
|
||||
if not setup_ids or len(setup_ids) != len(set(setup_ids)):
|
||||
raise LaboratorySetupRegistryError("setup IDs must be unique and non-empty")
|
||||
definition_versions = [
|
||||
(setup.run_definition.definition_id, setup.run_definition.version)
|
||||
for setup in self.setups
|
||||
if setup.run_definition is not None
|
||||
]
|
||||
if len(definition_versions) != len(set(definition_versions)):
|
||||
raise LaboratorySetupRegistryError("RunDefinition versions must be unique")
|
||||
result_ids = [
|
||||
result.result_id
|
||||
for setup in self.setups
|
||||
for result in setup.preserved_results
|
||||
]
|
||||
if len(result_ids) != len(set(result_ids)):
|
||||
raise LaboratorySetupRegistryError("preserved result IDs must be globally unique")
|
||||
|
||||
@classmethod
|
||||
def from_file(
|
||||
cls,
|
||||
path: Path,
|
||||
*,
|
||||
repository_root: Path,
|
||||
) -> LaboratorySetupRegistry:
|
||||
candidate = path.expanduser().absolute()
|
||||
root = repository_root.expanduser().resolve(strict=True)
|
||||
if candidate.is_symlink() or not candidate.is_file():
|
||||
raise LaboratorySetupRegistryError("setup registry must be a regular file")
|
||||
if candidate.stat().st_size > _MAX_REGISTRY_BYTES:
|
||||
raise LaboratorySetupRegistryError("setup registry is too large")
|
||||
try:
|
||||
document = _object(json.loads(candidate.read_text(encoding="utf-8")), "registry")
|
||||
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise LaboratorySetupRegistryError("setup registry is unreadable") from exc
|
||||
_exact_keys(document, {"schema_version", "setups"}, "registry")
|
||||
if document["schema_version"] != LABORATORY_SETUP_REGISTRY_SCHEMA:
|
||||
raise LaboratorySetupRegistryError("setup registry schema is invalid")
|
||||
rows = document["setups"]
|
||||
if not isinstance(rows, list):
|
||||
raise LaboratorySetupRegistryError("setup registry rows must be an array")
|
||||
return cls(tuple(_setup(row, repository_root=root) for row in rows))
|
||||
|
||||
def catalog(
|
||||
self,
|
||||
source: SessionSummary,
|
||||
*,
|
||||
available_observatory_result_ids: frozenset[str] = frozenset(),
|
||||
) -> dict[str, object]:
|
||||
return {
|
||||
"schema_version": LABORATORY_SETUP_CATALOG_SCHEMA,
|
||||
"source_session_id": source.session_id,
|
||||
"setups": [
|
||||
setup.project(
|
||||
source,
|
||||
available_observatory_result_ids=available_observatory_result_ids,
|
||||
)
|
||||
for setup in self.setups
|
||||
],
|
||||
"authority": dict(_AUTHORITY),
|
||||
}
|
||||
|
||||
def setup(self, setup_id: str) -> LaboratorySetup:
|
||||
for setup in self.setups:
|
||||
if setup.setup_id == setup_id:
|
||||
return setup
|
||||
raise KeyError(setup_id)
|
||||
|
||||
@property
|
||||
def observatory_result_ids(self) -> frozenset[str]:
|
||||
return frozenset(
|
||||
result.result_id
|
||||
for setup in self.setups
|
||||
for result in setup.preserved_results
|
||||
if result.access == "observatory"
|
||||
)
|
||||
|
||||
def observatory_result_kind(self, result_id: str) -> str | None:
|
||||
for setup in self.setups:
|
||||
for result in setup.preserved_results:
|
||||
if result.access == "observatory" and result.result_id == result_id:
|
||||
return result.result_kind
|
||||
return None
|
||||
|
||||
|
||||
def _setup(value: object, *, repository_root: Path) -> LaboratorySetup:
|
||||
row = _object(value, "setup")
|
||||
_exact_keys(
|
||||
row,
|
||||
{
|
||||
"setup_id",
|
||||
"display_name",
|
||||
"description",
|
||||
"origin",
|
||||
"source",
|
||||
"run_definition",
|
||||
"executor",
|
||||
"preserved_results",
|
||||
"authority",
|
||||
},
|
||||
"setup",
|
||||
)
|
||||
origin = row["origin"]
|
||||
if not isinstance(origin, str) or origin not in (
|
||||
"archived-definition",
|
||||
"existing-result",
|
||||
):
|
||||
raise LaboratorySetupRegistryError("setup origin is invalid")
|
||||
source = _object(row["source"], "source")
|
||||
_exact_keys(source, {"session_id", "label", "required_modalities"}, "source")
|
||||
modalities = _text_array(source["required_modalities"], "required_modalities")
|
||||
if not set(modalities).issubset(_MODALITIES):
|
||||
raise LaboratorySetupRegistryError("setup modalities are invalid")
|
||||
definition = (
|
||||
None
|
||||
if row["run_definition"] is None
|
||||
else _definition(row["run_definition"], repository_root=repository_root)
|
||||
)
|
||||
if (origin == "archived-definition") != (definition is not None):
|
||||
raise LaboratorySetupRegistryError("setup origin and RunDefinition disagree")
|
||||
if row["authority"] != _AUTHORITY:
|
||||
raise LaboratorySetupRegistryError("setup authority must remain observation-only")
|
||||
results = row["preserved_results"]
|
||||
if not isinstance(results, list) or not results:
|
||||
raise LaboratorySetupRegistryError("preserved results must be non-empty")
|
||||
preserved = tuple(_preserved_result(item) for item in results)
|
||||
if len({item.result_id for item in preserved}) != len(preserved):
|
||||
raise LaboratorySetupRegistryError("preserved result IDs must be unique")
|
||||
return LaboratorySetup(
|
||||
setup_id=_identifier(row["setup_id"], "setup_id"),
|
||||
display_name=_text(row["display_name"], "display_name"),
|
||||
description=_text(row["description"], "description"),
|
||||
origin=origin,
|
||||
source_session_id=_text(source["session_id"], "source session_id"),
|
||||
source_label=_text(source["label"], "source label"),
|
||||
required_modalities=modalities,
|
||||
run_definition=definition,
|
||||
executor=_executor(row["executor"]),
|
||||
preserved_results=preserved,
|
||||
)
|
||||
|
||||
|
||||
def _definition(value: object, *, repository_root: Path) -> _RunDefinition:
|
||||
row = _object(value, "RunDefinition")
|
||||
_exact_keys(
|
||||
row,
|
||||
{"definition_id", "version", "work_id", "configuration"},
|
||||
"RunDefinition",
|
||||
)
|
||||
version = row["version"]
|
||||
if not isinstance(version, int) or isinstance(version, bool) or version < 1:
|
||||
raise LaboratorySetupRegistryError("RunDefinition version is invalid")
|
||||
configuration = row["configuration"]
|
||||
if not isinstance(configuration, list) or not configuration:
|
||||
raise LaboratorySetupRegistryError("RunDefinition configuration is empty")
|
||||
references = tuple(
|
||||
_configuration(item, repository_root=repository_root) for item in configuration
|
||||
)
|
||||
if len({item.role for item in references}) != len(references):
|
||||
raise LaboratorySetupRegistryError("configuration roles must be unique")
|
||||
return _RunDefinition(
|
||||
definition_id=_identifier(row["definition_id"], "definition_id"),
|
||||
version=version,
|
||||
work_id=_identifier(row["work_id"], "work_id"),
|
||||
configuration=references,
|
||||
)
|
||||
|
||||
|
||||
def _configuration(value: object, *, repository_root: Path) -> _ConfigurationReference:
|
||||
row = _object(value, "configuration")
|
||||
_exact_keys(row, {"role", "path", "sha256"}, "configuration")
|
||||
relative = PurePosixPath(_text(row["path"], "configuration path"))
|
||||
if relative.is_absolute() or ".." in relative.parts:
|
||||
raise LaboratorySetupRegistryError("configuration path is unsafe")
|
||||
candidate = repository_root.joinpath(*relative.parts)
|
||||
if candidate.is_symlink() or not candidate.is_file():
|
||||
raise LaboratorySetupRegistryError("configuration file is unavailable")
|
||||
try:
|
||||
resolved = candidate.resolve(strict=True)
|
||||
if resolved.stat().st_size > _MAX_CONFIGURATION_BYTES:
|
||||
raise LaboratorySetupRegistryError("configuration file is too large")
|
||||
except OSError as exc:
|
||||
raise LaboratorySetupRegistryError("configuration file is unavailable") from exc
|
||||
if not resolved.is_relative_to(repository_root):
|
||||
raise LaboratorySetupRegistryError("configuration escaped repository root")
|
||||
digest = _digest(row["sha256"], "configuration sha256")
|
||||
if _file_sha256(resolved) != digest:
|
||||
raise LaboratorySetupRegistryError("configuration digest changed")
|
||||
return _ConfigurationReference(
|
||||
role=_identifier(row["role"], "configuration role"),
|
||||
path=relative,
|
||||
sha256=digest,
|
||||
)
|
||||
|
||||
|
||||
def _executor(value: object) -> _Executor:
|
||||
row = _object(value, "executor")
|
||||
_exact_keys(row, {"contour_id", "state", "reason_code", "reason"}, "executor")
|
||||
if row["state"] != "not-installed":
|
||||
raise LaboratorySetupRegistryError("only fail-closed executors are admitted in v1")
|
||||
return _Executor(
|
||||
contour_id=_identifier(row["contour_id"], "contour_id"),
|
||||
state="not-installed",
|
||||
reason_code=_identifier(row["reason_code"], "reason_code"),
|
||||
reason=_text(row["reason"], "executor reason"),
|
||||
)
|
||||
|
||||
|
||||
def _preserved_result(value: object) -> _PreservedResult:
|
||||
row = _object(value, "preserved result")
|
||||
_exact_keys(
|
||||
row,
|
||||
{"result_id", "result_kind", "relation", "access", "created_at_utc"},
|
||||
"preserved result",
|
||||
)
|
||||
result_id = _text(row["result_id"], "result_id")
|
||||
if _RESULT_ID.fullmatch(result_id) is None:
|
||||
raise LaboratorySetupRegistryError("result_id is invalid")
|
||||
access = row["access"]
|
||||
if not isinstance(access, str) or access not in (
|
||||
"legacy-lab",
|
||||
"observatory",
|
||||
"evidence-only",
|
||||
):
|
||||
raise LaboratorySetupRegistryError("result access is invalid")
|
||||
return _PreservedResult(
|
||||
result_id=result_id,
|
||||
result_kind=_identifier(row["result_kind"], "result_kind"),
|
||||
relation=_identifier(row["relation"], "result relation"),
|
||||
access=access,
|
||||
created_at_utc=_text(row["created_at_utc"], "created_at_utc"),
|
||||
)
|
||||
|
||||
|
||||
def _project_definition(
|
||||
setup: LaboratorySetup,
|
||||
definition: _RunDefinition | None,
|
||||
) -> dict[str, object] | None:
|
||||
if definition is None:
|
||||
return None
|
||||
identity = {
|
||||
"schema_version": "missioncore.observatory-run-definition/v1",
|
||||
"definition_id": definition.definition_id,
|
||||
"version": definition.version,
|
||||
"work_id": definition.work_id,
|
||||
"source": {
|
||||
"session_id": setup.source_session_id,
|
||||
"label": setup.source_label,
|
||||
"required_modalities": list(setup.required_modalities),
|
||||
},
|
||||
"configuration": [
|
||||
{"role": item.role, "sha256": item.sha256}
|
||||
for item in definition.configuration
|
||||
],
|
||||
"authority": dict(_AUTHORITY),
|
||||
}
|
||||
return {
|
||||
**identity,
|
||||
"definition_sha256": hashlib.sha256(_canonical_json(identity)).hexdigest(),
|
||||
}
|
||||
|
||||
|
||||
def _canonical_json(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _file_sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
try:
|
||||
with path.open("rb") as stream:
|
||||
while chunk := stream.read(128 * 1024):
|
||||
digest.update(chunk)
|
||||
except OSError as exc:
|
||||
raise LaboratorySetupRegistryError("configuration file is unavailable") from exc
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _reason(code: str, message: str) -> dict[str, str]:
|
||||
return {"code": code, "message": message}
|
||||
|
||||
|
||||
def _object(value: object, label: str) -> dict[str, Any]:
|
||||
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
|
||||
raise LaboratorySetupRegistryError(f"{label} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None:
|
||||
if set(value) != expected:
|
||||
raise LaboratorySetupRegistryError(f"{label} keys are invalid")
|
||||
|
||||
|
||||
def _text(value: object, label: str) -> str:
|
||||
if not isinstance(value, str) or not value.strip() or value != value.strip():
|
||||
raise LaboratorySetupRegistryError(f"{label} must be non-empty text")
|
||||
return value
|
||||
|
||||
|
||||
def _identifier(value: object, label: str) -> str:
|
||||
text = _text(value, label)
|
||||
if _IDENTIFIER.fullmatch(text) is None:
|
||||
raise LaboratorySetupRegistryError(f"{label} is invalid")
|
||||
return text
|
||||
|
||||
|
||||
def _digest(value: object, label: str) -> str:
|
||||
text = _text(value, label)
|
||||
if _SHA256.fullmatch(text) is None:
|
||||
raise LaboratorySetupRegistryError(f"{label} is invalid")
|
||||
return text
|
||||
|
||||
|
||||
def _text_array(value: object, label: str) -> tuple[str, ...]:
|
||||
if not isinstance(value, list) or not value:
|
||||
raise LaboratorySetupRegistryError(f"{label} must be a non-empty array")
|
||||
result = tuple(_text(item, label) for item in value)
|
||||
if len(result) != len(set(result)):
|
||||
raise LaboratorySetupRegistryError(f"{label} must contain unique values")
|
||||
return result
|
||||
Reference in New Issue
Block a user