nat.atof.io#

ATOF JSON-Lines I/O utilities.

Read and write ATOF event streams as JSON-Lines files (one JSON object per line).

Attributes#

`_event_adapter`
`_SCOPE_WIRE_ORDER`
`_MARK_WIRE_ORDER`

Functions#

`_reorder_for_wire`(→ dict)	Reorder a serialized event dict to match the ATOF spec §3 field order.
`read_jsonl`(→ list[nat.atof.events.Event])	Read an ATOF JSON-Lines file and return a list of typed Event objects.
`write_jsonl`(→ None)	Write a list of Event objects to a JSON-Lines file.

Module Contents#

_event_adapter#

_SCOPE_WIRE_ORDER = ('kind', 'scope_category', 'atof_version', 'category', 'category_profile', 'uuid',...#

_MARK_WIRE_ORDER = ('kind', 'atof_version', 'category', 'category_profile', 'uuid', 'parent_uuid', 'data',...#

_reorder_for_wire(event_dict: dict) → dict#

Reorder a serialized event dict to match the ATOF spec §3 field order.

Pydantic’s model_dump emits subclass fields after inherited base fields, which pushes kind/scope_category to the end of the output. The spec wire envelope example (§1) puts them first for readability. This reorders the dict while preserving any unknown keys (vendor extensions) at the end in insertion order.

read_jsonl(path: str | pathlib.Path) → list[nat.atof.events.Event]#

Read an ATOF JSON-Lines file and return a list of typed Event objects.

Each line is parsed as a JSON object and validated against the Event discriminated union. Blank lines are skipped. Events are returned sorted by .ts_micros (the normalized int-microsecond timestamp, spec §5.1) so downstream consumers get a stable ordering across mixed str/int timestamp streams.

write_jsonl( events: list[nat.atof.events.Event], path: str | pathlib.Path, ) → None#

Write a list of Event objects to a JSON-Lines file.

Each event is serialized as a single JSON line. The file ends with a trailing newline. Optional fields with None values are emitted as explicit null on the wire (matching the spec wire envelope example in atof-event-format.md §1).