686 lines
26 KiB
Python
686 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass, replace
|
|
from datetime import UTC, datetime
|
|
from enum import StrEnum
|
|
from pathlib import PurePosixPath
|
|
from typing import Any, Final
|
|
|
|
IDENTIFIER_PATTERN: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
|
SHA256_PATTERN: Final = re.compile(r"^[a-f0-9]{64}$")
|
|
GIT_REVISION_PATTERN: Final = re.compile(r"^[a-f0-9]{7,64}$")
|
|
|
|
|
|
class SimulationContractError(ValueError):
|
|
"""A Polygon domain object violates the admitted S1 contract."""
|
|
|
|
|
|
class RunKind(StrEnum):
|
|
SIMULATION_CLOSED_LOOP = "simulation_closed_loop"
|
|
REPLAY_SHADOW = "replay_shadow"
|
|
DIGITAL_TWIN_CLOSED_LOOP = "digital_twin_closed_loop"
|
|
CONTROLLER_IN_LOOP = "controller_in_loop"
|
|
HIL = "hil"
|
|
PHYSICAL_SHADOW = "physical_shadow"
|
|
|
|
|
|
class RunState(StrEnum):
|
|
ADMITTED = "admitted"
|
|
STARTING = "starting"
|
|
RUNNING = "running"
|
|
PAUSED = "paused"
|
|
STOPPING = "stopping"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
ABORTED = "aborted"
|
|
|
|
@property
|
|
def terminal(self) -> bool:
|
|
return self in {RunState.COMPLETED, RunState.FAILED, RunState.ABORTED}
|
|
|
|
|
|
class ReproducibilityTier(StrEnum):
|
|
R0 = "R0"
|
|
R1 = "R1"
|
|
R2 = "R2"
|
|
|
|
|
|
class ControlProfile(StrEnum):
|
|
ROVER_SPEED_STEERING_V1 = "rover-speed-steering/v1"
|
|
ROVER_SPEED_YAW_RATE_V1 = "rover-speed-yaw-rate/v1"
|
|
|
|
|
|
class CommandAuthorityScope(StrEnum):
|
|
VIRTUAL_ONLY = "virtual-only"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ProviderPin:
|
|
identifier: str
|
|
version: str
|
|
revision: str
|
|
digest: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.identifier, "provider identifier")
|
|
if not self.version.strip():
|
|
raise SimulationContractError("provider version must not be empty")
|
|
if not self.revision.strip():
|
|
raise SimulationContractError("provider revision must not be empty")
|
|
if self.digest is not None:
|
|
_sha256(self.digest, "provider digest")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"identifier": self.identifier,
|
|
"version": self.version,
|
|
"revision": self.revision,
|
|
"digest": self.digest,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> ProviderPin:
|
|
document = _object(value, "provider pin")
|
|
return cls(
|
|
identifier=_string(document, "identifier"),
|
|
version=_string(document, "version"),
|
|
revision=_string(document, "revision"),
|
|
digest=_optional_string(document, "digest"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AuthorityProfile:
|
|
"""The only command authority admitted before a physical safety gate."""
|
|
|
|
generation: int
|
|
command_ttl_max_ns: int
|
|
heartbeat_timeout_monotonic_ns: int
|
|
simulation_or_shadow_only: bool = True
|
|
actuator_authority: bool = False
|
|
navigation_or_safety_accepted: bool = False
|
|
direct_actuator_setpoints_allowed: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.generation < 1:
|
|
raise SimulationContractError("authority generation must be positive")
|
|
if self.command_ttl_max_ns < 1:
|
|
raise SimulationContractError("command TTL limit must be positive")
|
|
if self.heartbeat_timeout_monotonic_ns < 1:
|
|
raise SimulationContractError("heartbeat timeout must be positive")
|
|
if (
|
|
not self.simulation_or_shadow_only
|
|
or self.actuator_authority
|
|
or self.navigation_or_safety_accepted
|
|
or self.direct_actuator_setpoints_allowed
|
|
):
|
|
raise SimulationContractError(
|
|
"S1 authority must remain simulation/shadow-only without actuator authority"
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"generation": self.generation,
|
|
"command_ttl_max_ns": self.command_ttl_max_ns,
|
|
"heartbeat_timeout_monotonic_ns": self.heartbeat_timeout_monotonic_ns,
|
|
"simulation_or_shadow_only": self.simulation_or_shadow_only,
|
|
"actuator_authority": self.actuator_authority,
|
|
"navigation_or_safety_accepted": self.navigation_or_safety_accepted,
|
|
"direct_actuator_setpoints_allowed": self.direct_actuator_setpoints_allowed,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> AuthorityProfile:
|
|
document = _object(value, "authority profile")
|
|
return cls(
|
|
generation=_integer(document, "generation"),
|
|
command_ttl_max_ns=_integer(document, "command_ttl_max_ns"),
|
|
heartbeat_timeout_monotonic_ns=_integer(
|
|
document,
|
|
"heartbeat_timeout_monotonic_ns",
|
|
),
|
|
simulation_or_shadow_only=_boolean(document, "simulation_or_shadow_only"),
|
|
actuator_authority=_boolean(document, "actuator_authority"),
|
|
navigation_or_safety_accepted=_boolean(
|
|
document,
|
|
"navigation_or_safety_accepted",
|
|
),
|
|
direct_actuator_setpoints_allowed=_boolean(
|
|
document,
|
|
"direct_actuator_setpoints_allowed",
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class QualificationArtifact:
|
|
artifact_id: str
|
|
kind: str
|
|
relative_path: str
|
|
sha256: str
|
|
byte_length: int
|
|
source_of_record: bool
|
|
sequence: int = 0
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.artifact_id, "artifact id")
|
|
_identifier(self.kind, "artifact kind")
|
|
_relative_path(self.relative_path)
|
|
_sha256(self.sha256, "artifact digest")
|
|
if self.byte_length < 0:
|
|
raise SimulationContractError("artifact byte length must not be negative")
|
|
if self.sequence < 0:
|
|
raise SimulationContractError("artifact sequence must not be negative")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"artifact_id": self.artifact_id,
|
|
"kind": self.kind,
|
|
"relative_path": self.relative_path,
|
|
"sha256": self.sha256,
|
|
"byte_length": self.byte_length,
|
|
"source_of_record": self.source_of_record,
|
|
"sequence": self.sequence,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> QualificationArtifact:
|
|
document = _object(value, "qualification artifact")
|
|
return cls(
|
|
artifact_id=_string(document, "artifact_id"),
|
|
kind=_string(document, "kind"),
|
|
relative_path=_string(document, "relative_path"),
|
|
sha256=_string(document, "sha256"),
|
|
byte_length=_integer(document, "byte_length"),
|
|
source_of_record=_boolean(document, "source_of_record"),
|
|
sequence=_integer(document, "sequence"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class QualificationRun:
|
|
run_id: str
|
|
episode_id: str
|
|
kind: RunKind
|
|
state: RunState
|
|
scenario_generation: str
|
|
scenario_sha256: str
|
|
profile_generation: str
|
|
profile_sha256: str
|
|
mission_core_commit: str
|
|
providers: tuple[ProviderPin, ...]
|
|
host_profile_id: str
|
|
host_profile_sha256: str
|
|
seed: int
|
|
reproducibility_tier: ReproducibilityTier
|
|
authority: AuthorityProfile
|
|
clock_domain: str
|
|
created_at_utc: str
|
|
started_at_utc: str | None = None
|
|
ended_at_utc: str | None = None
|
|
terminal_reason: str | None = None
|
|
parent_run_id: str | None = None
|
|
revision: int = 0
|
|
artifacts: tuple[QualificationArtifact, ...] = ()
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.run_id, "run id")
|
|
_identifier(self.episode_id, "episode id")
|
|
_identifier(self.scenario_generation, "scenario generation")
|
|
_sha256(self.scenario_sha256, "scenario digest")
|
|
_identifier(self.profile_generation, "profile generation")
|
|
_sha256(self.profile_sha256, "profile digest")
|
|
if not GIT_REVISION_PATTERN.fullmatch(self.mission_core_commit):
|
|
raise SimulationContractError("Mission Core commit must be a hexadecimal revision")
|
|
if not self.providers:
|
|
raise SimulationContractError("at least one provider pin is required")
|
|
provider_ids = [provider.identifier for provider in self.providers]
|
|
if len(provider_ids) != len(set(provider_ids)):
|
|
raise SimulationContractError("provider identifiers must be unique")
|
|
_identifier(self.host_profile_id, "host profile id")
|
|
_sha256(self.host_profile_sha256, "host profile digest")
|
|
if self.seed < 0:
|
|
raise SimulationContractError("run seed must not be negative")
|
|
if self.clock_domain != "gazebo:/clock" and self.kind in {
|
|
RunKind.SIMULATION_CLOSED_LOOP,
|
|
RunKind.DIGITAL_TWIN_CLOSED_LOOP,
|
|
}:
|
|
raise SimulationContractError("closed-loop simulation requires Gazebo /clock")
|
|
_utc_timestamp(self.created_at_utc, "created timestamp")
|
|
if self.started_at_utc is not None:
|
|
_utc_timestamp(self.started_at_utc, "started timestamp")
|
|
if self.ended_at_utc is not None:
|
|
_utc_timestamp(self.ended_at_utc, "ended timestamp")
|
|
if self.parent_run_id is not None:
|
|
_identifier(self.parent_run_id, "parent run id")
|
|
if self.parent_run_id == self.run_id:
|
|
raise SimulationContractError("a run cannot be its own parent")
|
|
if self.revision < 0:
|
|
raise SimulationContractError("run revision must not be negative")
|
|
if self.state.terminal != (self.ended_at_utc is not None):
|
|
raise SimulationContractError("terminal state and ended timestamp must agree")
|
|
if self.state.terminal != (self.terminal_reason is not None):
|
|
raise SimulationContractError("terminal state and terminal reason must agree")
|
|
if self.kind in {RunKind.CONTROLLER_IN_LOOP, RunKind.HIL}:
|
|
raise SimulationContractError(
|
|
f"{self.kind.value} requires a separate physical authority safety gate"
|
|
)
|
|
|
|
def with_runtime(
|
|
self,
|
|
*,
|
|
state: RunState,
|
|
revision: int,
|
|
started_at_utc: str | None,
|
|
ended_at_utc: str | None,
|
|
terminal_reason: str | None,
|
|
artifacts: tuple[QualificationArtifact, ...],
|
|
) -> QualificationRun:
|
|
return replace(
|
|
self,
|
|
state=state,
|
|
revision=revision,
|
|
started_at_utc=started_at_utc,
|
|
ended_at_utc=ended_at_utc,
|
|
terminal_reason=terminal_reason,
|
|
artifacts=artifacts,
|
|
)
|
|
|
|
def to_manifest_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "missioncore.qualification-run/v1",
|
|
"run_id": self.run_id,
|
|
"episode_id": self.episode_id,
|
|
"kind": self.kind.value,
|
|
"initial_state": RunState.ADMITTED.value,
|
|
"scenario": {
|
|
"generation": self.scenario_generation,
|
|
"sha256": self.scenario_sha256,
|
|
},
|
|
"profile": {
|
|
"generation": self.profile_generation,
|
|
"sha256": self.profile_sha256,
|
|
},
|
|
"mission_core_commit": self.mission_core_commit,
|
|
"providers": [provider.to_dict() for provider in self.providers],
|
|
"host_profile": {
|
|
"id": self.host_profile_id,
|
|
"sha256": self.host_profile_sha256,
|
|
},
|
|
"seed": self.seed,
|
|
"reproducibility_tier": self.reproducibility_tier.value,
|
|
"authority": self.authority.to_dict(),
|
|
"clock_domain": self.clock_domain,
|
|
"created_at_utc": self.created_at_utc,
|
|
"parent_run_id": self.parent_run_id,
|
|
}
|
|
|
|
@classmethod
|
|
def from_manifest_dict(cls, value: object) -> QualificationRun:
|
|
document = _object(value, "qualification run")
|
|
if document.get("schema_version") != "missioncore.qualification-run/v1":
|
|
raise SimulationContractError("unsupported qualification-run schema")
|
|
if document.get("initial_state") != RunState.ADMITTED.value:
|
|
raise SimulationContractError("qualification run must be admitted immutably")
|
|
scenario = _object(document.get("scenario"), "scenario")
|
|
profile = _object(document.get("profile"), "profile")
|
|
host_profile = _object(document.get("host_profile"), "host profile")
|
|
providers = document.get("providers")
|
|
if not isinstance(providers, list):
|
|
raise SimulationContractError("providers must be an array")
|
|
try:
|
|
kind = RunKind(_string(document, "kind"))
|
|
tier = ReproducibilityTier(_string(document, "reproducibility_tier"))
|
|
except ValueError as exc:
|
|
raise SimulationContractError("qualification run contains an unknown enum") from exc
|
|
return cls(
|
|
run_id=_string(document, "run_id"),
|
|
episode_id=_string(document, "episode_id"),
|
|
kind=kind,
|
|
state=RunState.ADMITTED,
|
|
scenario_generation=_string(scenario, "generation"),
|
|
scenario_sha256=_string(scenario, "sha256"),
|
|
profile_generation=_string(profile, "generation"),
|
|
profile_sha256=_string(profile, "sha256"),
|
|
mission_core_commit=_string(document, "mission_core_commit"),
|
|
providers=tuple(ProviderPin.from_dict(provider) for provider in providers),
|
|
host_profile_id=_string(host_profile, "id"),
|
|
host_profile_sha256=_string(host_profile, "sha256"),
|
|
seed=_integer(document, "seed"),
|
|
reproducibility_tier=tier,
|
|
authority=AuthorityProfile.from_dict(document.get("authority")),
|
|
clock_domain=_string(document, "clock_domain"),
|
|
created_at_utc=_string(document, "created_at_utc"),
|
|
parent_run_id=_optional_string(document, "parent_run_id"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class QualificationEvent:
|
|
run_id: str
|
|
sequence: int
|
|
event_type: str
|
|
observed_at_utc: str
|
|
host_monotonic_ns: int
|
|
sim_time_ns: int | None
|
|
payload: dict[str, Any]
|
|
|
|
def __post_init__(self) -> None:
|
|
_identifier(self.run_id, "event run id")
|
|
_identifier(self.event_type, "event type")
|
|
if self.sequence < 1:
|
|
raise SimulationContractError("event sequence must be positive")
|
|
_utc_timestamp(self.observed_at_utc, "event timestamp")
|
|
if self.host_monotonic_ns < 0:
|
|
raise SimulationContractError("host monotonic timestamp must not be negative")
|
|
if self.sim_time_ns is not None and self.sim_time_ns < 0:
|
|
raise SimulationContractError("simulation timestamp must not be negative")
|
|
_json_object(self.payload, "event payload")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "missioncore.qualification-event/v1",
|
|
"run_id": self.run_id,
|
|
"sequence": self.sequence,
|
|
"event_type": self.event_type,
|
|
"observed_at_utc": self.observed_at_utc,
|
|
"host_monotonic_ns": self.host_monotonic_ns,
|
|
"sim_time_ns": self.sim_time_ns,
|
|
"payload": self.payload,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> QualificationEvent:
|
|
document = _object(value, "qualification event")
|
|
if document.get("schema_version") != "missioncore.qualification-event/v1":
|
|
raise SimulationContractError("unsupported qualification-event schema")
|
|
sim_time = document.get("sim_time_ns")
|
|
if sim_time is not None and (not isinstance(sim_time, int) or isinstance(sim_time, bool)):
|
|
raise SimulationContractError("sim_time_ns must be an integer or null")
|
|
payload = _object(document.get("payload"), "event payload")
|
|
return cls(
|
|
run_id=_string(document, "run_id"),
|
|
sequence=_integer(document, "sequence"),
|
|
event_type=_string(document, "event_type"),
|
|
observed_at_utc=_string(document, "observed_at_utc"),
|
|
host_monotonic_ns=_integer(document, "host_monotonic_ns"),
|
|
sim_time_ns=sim_time,
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AckermannControlSetpoint:
|
|
run_id: str
|
|
command_id: str
|
|
sequence: int
|
|
issued_at_sim_ns: int
|
|
valid_until_sim_ns: int
|
|
authority_generation: int
|
|
speed_mps: float
|
|
steering_normalized: float
|
|
profile: ControlProfile = ControlProfile.ROVER_SPEED_STEERING_V1
|
|
authority_scope: CommandAuthorityScope = CommandAuthorityScope.VIRTUAL_ONLY
|
|
source: str = "simulation-orchestrator"
|
|
|
|
def __post_init__(self) -> None:
|
|
_command_header(
|
|
self.run_id,
|
|
self.command_id,
|
|
self.sequence,
|
|
self.issued_at_sim_ns,
|
|
self.valid_until_sim_ns,
|
|
self.authority_generation,
|
|
)
|
|
_finite(self.speed_mps, "speed")
|
|
_finite(self.steering_normalized, "steering")
|
|
if not -1.0 <= self.steering_normalized <= 1.0:
|
|
raise SimulationContractError("normalized steering must be within -1..1")
|
|
if self.profile is not ControlProfile.ROVER_SPEED_STEERING_V1:
|
|
raise SimulationContractError("Ackermann command has an incompatible profile")
|
|
_command_provenance(self.authority_scope, self.source)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "missioncore.control-setpoint/v1",
|
|
"run_id": self.run_id,
|
|
"command_id": self.command_id,
|
|
"sequence": self.sequence,
|
|
"profile": self.profile.value,
|
|
"authority_scope": self.authority_scope.value,
|
|
"source": self.source,
|
|
"issued_at_sim_ns": self.issued_at_sim_ns,
|
|
"valid_until_sim_ns": self.valid_until_sim_ns,
|
|
"authority_generation": self.authority_generation,
|
|
"speed_mps": self.speed_mps,
|
|
"steering_normalized": self.steering_normalized,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DifferentialControlSetpoint:
|
|
run_id: str
|
|
command_id: str
|
|
sequence: int
|
|
issued_at_sim_ns: int
|
|
valid_until_sim_ns: int
|
|
authority_generation: int
|
|
speed_mps: float
|
|
yaw_rate_rps: float
|
|
profile: ControlProfile = ControlProfile.ROVER_SPEED_YAW_RATE_V1
|
|
authority_scope: CommandAuthorityScope = CommandAuthorityScope.VIRTUAL_ONLY
|
|
source: str = "simulation-orchestrator"
|
|
|
|
def __post_init__(self) -> None:
|
|
_command_header(
|
|
self.run_id,
|
|
self.command_id,
|
|
self.sequence,
|
|
self.issued_at_sim_ns,
|
|
self.valid_until_sim_ns,
|
|
self.authority_generation,
|
|
)
|
|
_finite(self.speed_mps, "speed")
|
|
_finite(self.yaw_rate_rps, "yaw rate")
|
|
if self.profile is not ControlProfile.ROVER_SPEED_YAW_RATE_V1:
|
|
raise SimulationContractError("differential command has an incompatible profile")
|
|
_command_provenance(self.authority_scope, self.source)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "missioncore.control-setpoint/v1",
|
|
"run_id": self.run_id,
|
|
"command_id": self.command_id,
|
|
"sequence": self.sequence,
|
|
"profile": self.profile.value,
|
|
"authority_scope": self.authority_scope.value,
|
|
"source": self.source,
|
|
"issued_at_sim_ns": self.issued_at_sim_ns,
|
|
"valid_until_sim_ns": self.valid_until_sim_ns,
|
|
"authority_generation": self.authority_generation,
|
|
"speed_mps": self.speed_mps,
|
|
"yaw_rate_rps": self.yaw_rate_rps,
|
|
}
|
|
|
|
|
|
ControlSetpoint = AckermannControlSetpoint | DifferentialControlSetpoint
|
|
|
|
|
|
def control_setpoint_from_dict(value: object) -> ControlSetpoint:
|
|
document = _object(value, "control setpoint")
|
|
if document.get("schema_version") != "missioncore.control-setpoint/v1":
|
|
raise SimulationContractError("unsupported control-setpoint schema")
|
|
try:
|
|
profile = ControlProfile(_string(document, "profile"))
|
|
authority_scope = CommandAuthorityScope(_string(document, "authority_scope"))
|
|
except ValueError as exc:
|
|
raise SimulationContractError("unknown control profile or authority scope") from exc
|
|
run_id = _string(document, "run_id")
|
|
command_id = _string(document, "command_id")
|
|
sequence = _integer(document, "sequence")
|
|
issued_at_sim_ns = _integer(document, "issued_at_sim_ns")
|
|
valid_until_sim_ns = _integer(document, "valid_until_sim_ns")
|
|
authority_generation = _integer(document, "authority_generation")
|
|
speed_mps = _number(document, "speed_mps")
|
|
if profile is ControlProfile.ROVER_SPEED_STEERING_V1:
|
|
return AckermannControlSetpoint(
|
|
run_id=run_id,
|
|
command_id=command_id,
|
|
sequence=sequence,
|
|
issued_at_sim_ns=issued_at_sim_ns,
|
|
valid_until_sim_ns=valid_until_sim_ns,
|
|
authority_generation=authority_generation,
|
|
speed_mps=speed_mps,
|
|
steering_normalized=_number(document, "steering_normalized"),
|
|
authority_scope=authority_scope,
|
|
source=_string(document, "source"),
|
|
)
|
|
return DifferentialControlSetpoint(
|
|
run_id=run_id,
|
|
command_id=command_id,
|
|
sequence=sequence,
|
|
issued_at_sim_ns=issued_at_sim_ns,
|
|
valid_until_sim_ns=valid_until_sim_ns,
|
|
authority_generation=authority_generation,
|
|
speed_mps=speed_mps,
|
|
yaw_rate_rps=_number(document, "yaw_rate_rps"),
|
|
authority_scope=authority_scope,
|
|
source=_string(document, "source"),
|
|
)
|
|
|
|
|
|
def _command_header(
|
|
run_id: str,
|
|
command_id: str,
|
|
sequence: int,
|
|
issued_at_sim_ns: int,
|
|
valid_until_sim_ns: int,
|
|
authority_generation: int,
|
|
) -> None:
|
|
_identifier(run_id, "command run id")
|
|
_identifier(command_id, "command id")
|
|
if sequence < 1:
|
|
raise SimulationContractError("command sequence must be positive")
|
|
if issued_at_sim_ns < 0:
|
|
raise SimulationContractError("command issue time must not be negative")
|
|
if valid_until_sim_ns <= issued_at_sim_ns:
|
|
raise SimulationContractError("command validity deadline must follow issue time")
|
|
if authority_generation < 1:
|
|
raise SimulationContractError("command authority generation must be positive")
|
|
|
|
|
|
def _command_provenance(authority_scope: CommandAuthorityScope, source: str) -> None:
|
|
if authority_scope is not CommandAuthorityScope.VIRTUAL_ONLY:
|
|
raise SimulationContractError("S1 commands require virtual-only authority scope")
|
|
_identifier(source, "command source")
|
|
|
|
|
|
def _identifier(value: str, label: str) -> str:
|
|
if not IDENTIFIER_PATTERN.fullmatch(value):
|
|
raise SimulationContractError(f"{label} is not a safe identifier")
|
|
return value
|
|
|
|
|
|
def _sha256(value: str, label: str) -> str:
|
|
if not SHA256_PATTERN.fullmatch(value):
|
|
raise SimulationContractError(f"{label} must be a lowercase SHA-256 digest")
|
|
return value
|
|
|
|
|
|
def _relative_path(value: str) -> str:
|
|
path = PurePosixPath(value)
|
|
if (
|
|
not value
|
|
or value.startswith("/")
|
|
or "\\" in value
|
|
or path.is_absolute()
|
|
or path.as_posix() != value
|
|
or any(part in {"", ".", ".."} for part in path.parts)
|
|
):
|
|
raise SimulationContractError("artifact path must be a confined relative POSIX path")
|
|
return value
|
|
|
|
|
|
def _utc_timestamp(value: str, label: str) -> str:
|
|
if not value.endswith("Z") or "T" not in value:
|
|
raise SimulationContractError(f"{label} must be an explicit UTC timestamp")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.removesuffix("Z") + "+00:00")
|
|
except ValueError as exc:
|
|
raise SimulationContractError(f"{label} must be a valid explicit UTC timestamp") from exc
|
|
if parsed.tzinfo != UTC:
|
|
raise SimulationContractError(f"{label} must be an explicit UTC timestamp")
|
|
return value
|
|
|
|
|
|
def _finite(value: float, label: str) -> float:
|
|
if not math.isfinite(value):
|
|
raise SimulationContractError(f"{label} must be finite")
|
|
return value
|
|
|
|
|
|
def _object(value: object, label: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
|
|
raise SimulationContractError(f"{label} must be an object")
|
|
return value
|
|
|
|
|
|
def _json_object(value: object, label: str) -> dict[str, Any]:
|
|
document = _object(value, label)
|
|
_json_value(document, label)
|
|
return document
|
|
|
|
|
|
def _json_value(value: object, label: str) -> None:
|
|
if value is None or isinstance(value, str | bool | int):
|
|
return
|
|
if isinstance(value, float):
|
|
_finite(value, label)
|
|
return
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
_json_value(item, label)
|
|
return
|
|
if isinstance(value, dict) and all(isinstance(key, str) for key in value):
|
|
for item in value.values():
|
|
_json_value(item, label)
|
|
return
|
|
raise SimulationContractError(f"{label} must contain only finite JSON values")
|
|
|
|
|
|
def _string(document: dict[str, Any], key: str) -> str:
|
|
value = document.get(key)
|
|
if not isinstance(value, str):
|
|
raise SimulationContractError(f"{key} must be a string")
|
|
return value
|
|
|
|
|
|
def _optional_string(document: dict[str, Any], key: str) -> str | None:
|
|
value = document.get(key)
|
|
if value is not None and not isinstance(value, str):
|
|
raise SimulationContractError(f"{key} must be a string or null")
|
|
return value
|
|
|
|
|
|
def _integer(document: dict[str, Any], key: str) -> int:
|
|
value = document.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise SimulationContractError(f"{key} must be an integer")
|
|
return value
|
|
|
|
|
|
def _number(document: dict[str, Any], key: str) -> float:
|
|
value = document.get(key)
|
|
if not isinstance(value, int | float) or isinstance(value, bool):
|
|
raise SimulationContractError(f"{key} must be a number")
|
|
return float(value)
|
|
|
|
|
|
def _boolean(document: dict[str, Any], key: str) -> bool:
|
|
value = document.get(key)
|
|
if not isinstance(value, bool):
|
|
raise SimulationContractError(f"{key} must be a boolean")
|
|
return value
|