File size: 4,631 Bytes
767a051 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """Command-line interface for TOM World & Query Kernel 0.3 interval events."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
from .baseline import trusted_affine_baseline
from .canonical import canonical_bytes
from .interval import ClosedInterval
from .io import load_world
from .rational import Q
from .solver import certify_crossing, events_certificate, next_event_set
from .transitions import apply_event_set
def _emit(value: Any, output: str | None) -> None:
data = canonical_bytes(value) + b"\n"
if output:
path = Path(output)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
print(json.dumps(value, indent=2, sort_keys=True))
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="tom-world03", description="Certified rational interval-event queries")
sub = parser.add_subparsers(dest="command", required=True)
p_validate = sub.add_parser("validate-world")
p_validate.add_argument("world")
p_validate.add_argument("--output")
p_certify = sub.add_parser("certify-crossing")
p_certify.add_argument("world")
p_certify.add_argument("relation")
p_certify.add_argument("lower")
p_certify.add_argument("upper")
p_certify.add_argument("--refine-steps", type=int)
p_certify.add_argument("--output")
p_events = sub.add_parser("events")
p_events.add_argument("world")
p_events.add_argument("start")
p_events.add_argument("end")
p_events.add_argument("--output")
p_next = sub.add_parser("next-event-set")
p_next.add_argument("world")
p_next.add_argument("after")
p_next.add_argument("before")
p_next.add_argument("--include-after", action="store_true")
p_next.add_argument("--output")
p_apply = sub.add_parser("apply-next-event-set")
p_apply.add_argument("world")
p_apply.add_argument("after")
p_apply.add_argument("before")
p_apply.add_argument("--output")
p_baseline = sub.add_parser("compare-baseline")
p_baseline.add_argument("world")
p_baseline.add_argument("start")
p_baseline.add_argument("end")
p_baseline.add_argument("--output")
args = parser.parse_args(argv)
record, world = load_world(args.world)
if args.command == "validate-world":
_emit({
"schema": "TOM-WORLD-VALIDATION-0.3",
"status": "valid",
"world_hash": world.content_hash,
"trajectory": world.trajectory.id,
"relations": len(world.relations),
}, args.output)
return 0
if args.command == "certify-crossing":
relation = next((r for r in world.relations if r.id == args.relation), None)
if relation is None:
raise SystemExit(f"unknown relation {args.relation}")
result = certify_crossing(
world, relation, ClosedInterval(Q.from_value(args.lower), Q.from_value(args.upper)),
refine_steps=args.refine_steps,
)
_emit(result.certificate, args.output)
return 0
if args.command == "events":
_emit(events_certificate(world, args.start, args.end), args.output)
return 0
if args.command == "next-event-set":
_emit(next_event_set(world, args.after, args.before, include_after=args.include_after), args.output)
return 0
if args.command == "apply-next-event-set":
event_set = next_event_set(world, args.after, args.before)
_emit(apply_event_set(world, event_set), args.output)
return 0
if args.command == "compare-baseline":
baseline = trusted_affine_baseline(record, args.start, args.end)
solver = events_certificate(world, args.start, args.end)
solver_events = [
{
"relation_id": event["relation_id"],
"event_id": event["event_id"],
"priority": event["priority"],
"root": event["exact_root_time"],
}
for event in solver["events"] if event["exact_root_time"] is not None
]
result = {
"schema": "TOM-INTERVAL-BASELINE-COMPARISON-0.3",
"world_hash": world.content_hash,
"equal": solver_events == baseline["events"],
"solver_events": solver_events,
"baseline_events": baseline["events"],
"baseline_hash": baseline["content_hash"],
"solver_hash": solver["content_hash"],
}
_emit(result, args.output)
return 0
return 2
if __name__ == "__main__":
raise SystemExit(main())
|