Skip to content

lunartrace.science.ode

Lunar ODE source custody: fetch pages, retain their bytes, and read them back offline.

Moved out of tools/vikram_study.py so the Python facade and the tool share one implementation. The method itself lives in lunartrace.science.temporal and is not changed here.

A retained source is the exact bytes of every ODE response plus a byte-level receipt (fetch-receipt.json) that names each response file, its URL, headers, size and SHA-256. Retained holds those bytes in memory. load_retained reads them from a directory (or a packaged snapshot), write_retained puts them back on disk, and source_record checks every page against its receipt digest before it returns the source object that the Study record embeds.

Only fetch_pages and fetch touch the network.

class

Retained

The retained bytes of one fetch: the receipt file and every raw response file.

files maps the receipt’s relative names (raw/page-000.json) to their exact bytes.

The parsed receipt.

function

def http_get(url: str) -> tuple[int, dict[str, str], bytes]

GET url; return the status, the kept response headers and the body bytes.

function

def fetch_pages(query: ODEQuery, *, get: Getter = http_get, pause_s: float = 1.0) -> Retained

Query Lunar ODE now and keep every raw response in memory with a byte-level receipt.

function

def write_retained(retained: Retained, out: Path) -> None

Write the receipt and every raw file under out exactly as retained.

function

def fetch(query: ODEQuery, out: Path) -> dict[str, Any]

Query Lunar ODE now and retain every raw page plus the receipt under out.

function

def load_retained(root: Path | Traversable) -> Retained

Read a retained fetch from a directory or a packaged snapshot.

  • FileNotFoundError — If the receipt or a file it names is missing.

function

def source_record(retained: Retained) -> dict[str, Any]

The source object of a Study, after checking every file against its receipt digest.

  • ValueError — If any retained file no longer matches the SHA-256 the receipt recorded for it.

function

def rows(retained: Retained) -> list[Observation]

Parse every retained raw/page-*.json into observation rows, in page order.

function

def load_rows(out: Path) -> list[Observation]

Observation rows of the retained fetch stored under the directory out.

function

def load_source_record(out: Path) -> dict[str, Any]

The checked source object of the retained fetch stored under out.