Skip to content

lpgcast.output

A cast on disk.

File Holds Same bytes on replay
graph.json the schema and the graph (graflo GraFloOutput) yes
groundings.jsonl one grounding per vertex, edge and property yes
pending.jsonl edges awaiting a secondary-identity lookup yes
episodes.jsonl one result per episode yes
quarantine.jsonl what was left out, and unmapped items yes
manifest.json versions, settings, counts and usage no: usage differs
write.json what was sent to the database, when one was given —

manifest.json names the output_format_version. Version 2: a grounding whose in_context is set has offsets into that context item's text.

Attributes

QUARANTINE_FILE = 'quarantine.jsonl' module-attribute

Classes

Functions:

save_cast(result, out_dir, *, write_report=None)

Write result under out_dir (created if needed); returns the directory.

Source code in lpgcast/output.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def save_cast(
    result: CastResult, out_dir: str | Path, *, write_report: BaseModel | None = None
) -> Path:
    """Write *result* under *out_dir* (created if needed); returns the directory."""
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    (out / "graph.json").write_text(
        _json(result.output.to_dict()) + "\n", encoding="utf-8"
    )
    (out / "groundings.jsonl").write_text(_jsonl(result.groundings), encoding="utf-8")
    (out / "pending.jsonl").write_text(_jsonl(result.pending), encoding="utf-8")
    (out / "episodes.jsonl").write_text(_jsonl(result.episodes), encoding="utf-8")
    (out / "manifest.json").write_text(
        _json(result.manifest.model_dump(mode="json")) + "\n", encoding="utf-8"
    )
    if write_report is not None:
        (out / "write.json").write_text(
            _json(write_report.model_dump(mode="json")) + "\n", encoding="utf-8"
        )
    return out