Skip to content

Python API

Kind: reference.

This page renders the experiment's Python docstrings and signatures from source. These interfaces support the current experiment. They do not promise a stable public API. Use Commands and scoring to run the receipt workflow.

The modules below cover receipt metrics, inference requests, synthetic sweeps, synthetic inputs, container ownership, optional service handover and CLI output. The source docstrings define each function's behavior and limits.

inference_lab.receipt_metrics

Measure diagnostic overlap against partial CORD annotations.

These metrics are not official CORD scores. They do not assess layout or associate extracted amounts with financial fields.

Attributes:

Name Type Description
TOKEN

Pattern preserving numeric punctuation while extracting word tokens.
CONTROL

Pattern matching known Nemotron control tokens.

Examples:

from inference_lab.receipt_metrics import score

recall = score("Total 12.00", "total 12.00")["numeric_recall"]  # 1.0
See Also

inference_lab.client: HTTP extraction and request status accounting.

numbers(text: str) -> Counter

Count exact numeric strings, including repeated occurrences.

Parameters:

Name Type Description Default
text str

Reference or predicted text.

 required

Returns:

Type Description
Counter

Counts keyed by numeric token; punctuation and signs remain significant.
Source code in src/inference_lab/receipt_metrics.py 
59
60
61
62
63
64
65
66
67
68
def numbers(text: str) -> Counter:
    """Count exact numeric strings, including repeated occurrences.

    Args:
        text: Reference or predicted text.

    Returns:
        Counts keyed by numeric token; punctuation and signs remain significant.
    """
    return Counter(t for t in tokens(text) if re.fullmatch(r"[+-]?\d[\d.,:/%-]*", t))

plain_text(raw: str) -> str

Strip model markup and collapse whitespace for diagnostic scoring.

Parameters:

Name Type Description Default
raw str

Model output containing control tokens, markup or LaTeX commands.

 required

Returns:

Type Description
str

Unescaped text with markup removed and whitespace collapsed.
Source code in src/inference_lab/receipt_metrics.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def plain_text(raw: str) -> str:
    """Strip model markup and collapse whitespace for diagnostic scoring.

    Args:
        raw: Model output containing control tokens, markup or LaTeX commands.

    Returns:
        Unescaped text with markup removed and whitespace collapsed.
    """
    text = CONTROL.sub(" ", raw)
    text = re.sub(r"<[^>]+>", " ", text)
    text = re.sub(r"\\(?:begin|end)\{[^}]+\}(?:\{[^}]+\})?", " ", text)
    text = re.sub(r"\\multicolumn\{[^}]*\}\{[^}]*\}", " ", text)
    text = re.sub(r"\\[A-Za-z]+\*?", " ", text)
    return " ".join(html.unescape(text).split())

score(reference: str, prediction: str) -> dict

Calculate token ROUGE-L and exact numeric-string recall.

Parameters:

Name Type Description Default
reference str

Partial annotated text used as the comparison target.

 required
prediction str

Extracted plain text to compare with the reference.

 required

Returns:

Type Description
dict

Token precision, recall and F1; numeric match counts and recall; and
dict

unmatched reference numeric strings. Recall and F1 are None for an
dict

empty reference. Numeric recall is None when it contains no numbers.
Source code in src/inference_lab/receipt_metrics.py 
 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
def score(reference: str, prediction: str) -> dict:
    """Calculate token ROUGE-L and exact numeric-string recall.

    Args:
        reference: Partial annotated text used as the comparison target.
        prediction: Extracted plain text to compare with the reference.

    Returns:
        Token precision, recall and F1; numeric match counts and recall; and
        unmatched reference numeric strings. Recall and F1 are None for an
        empty reference. Numeric recall is None when it contains no numbers.
    """
    ref, pred = tokens(reference), tokens(prediction)
    previous = [0] * (len(pred) + 1)
    for token in ref:
        row = [0]
        for j, other in enumerate(pred, 1):
            row.append(
                previous[j - 1] + 1 if token == other else max(previous[j], row[-1])
            )
        previous = row
    overlap = previous[-1]
    precision = overlap / len(pred) if pred else 0.0
    recall = overlap / len(ref) if ref else None
    f1 = 2 * overlap / (len(ref) + len(pred)) if ref else None
    rn, pn = numbers(reference), numbers(prediction)
    matched = sum((rn & pn).values())
    count = sum(rn.values())
    return {
        "rouge_l_precision": precision,
        "rouge_l_recall": recall,
        "rouge_l_f1": f1,
        "reference_tokens": len(ref),
        "prediction_tokens": len(pred),
        "numeric_reference_count": count,
        "numeric_matched_count": matched,
        "numeric_recall": matched / count if count else None,
        "missing_numbers": list((rn - pn).elements()),
    }

tokens(text: str) -> list[str]

Extract casefolded word and numeric tokens.

Parameters:

Name Type Description Default
text str

Reference or predicted text.

 required

Returns:

Type Description
list[str]

Tokens in source order, with numeric punctuation preserved.
Source code in src/inference_lab/receipt_metrics.py 
47
48
49
50
51
52
53
54
55
56
def tokens(text: str) -> list[str]:
    """Extract casefolded word and numeric tokens.

    Args:
        text: Reference or predicted text.

    Returns:
        Tokens in source order, with numeric punctuation preserved.
    """
    return TOKEN.findall(text.casefold())

inference_lab.client

Send one prepared image per bounded HTTP request without retries.

Attributes:

Name Type Description
PROMPT

Control tokens for Nemotron layout and text extraction.
MODEL

Model name exposed by the demo server.

Examples:

Normalize expected text for a case-insensitive sanity check:

from inference_lab.client import normalized

text = normalized("Total: 12.00")  # "total 12 00"
See Also

inference_lab.experiment: Preparation and measured concurrency sweeps.

Page dataclass

Hold an image prepared outside the measured request interval.

Attributes:

Name Type Description
name str

Source image filename used in request records.
sha256 str

SHA-256 digest of the original file bytes.
data_url str

Base64 PNG data URL submitted to the server.
expected tuple[str, ...]

Text snippets required by the synthetic quality check.

Examples:

Prepare a real image rather than constructing its data URL manually:

from pathlib import Path
from inference_lab.client import prepare

page = prepare(Path("corpus/page-01.png"), ("LAB401",))
Source code in src/inference_lab/client.py 
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
@dataclass(frozen=True)
class Page:
    """Hold an image prepared outside the measured request interval.

    Attributes:
        name (str): Source image filename used in request records.
        sha256 (str): SHA-256 digest of the original file bytes.
        data_url (str): Base64 PNG data URL submitted to the server.
        expected (tuple[str, ...]): Text snippets required by the synthetic quality check.

    Examples:
        Prepare a real image rather than constructing its data URL manually:

        ```python
        from pathlib import Path
        from inference_lab.client import prepare

        page = prepare(Path("corpus/page-01.png"), ("LAB401",))
        ```
    """

    name: str
    sha256: str
    data_url: str
    expected: tuple[str, ...]

infer(client: httpx.AsyncClient, page: Page, timeout: float) -> dict async

Return one terminal record, including failure and observed latency.

HTTP, timeout and malformed-response failures become records rather than escaping to the caller. Task cancellation still propagates.

Parameters:

Name Type Description Default
client AsyncClient

Async HTTP client configured with the server base URL.

 required
page Page

Image and expected text prepared before timing.

 required
timeout float

Total request deadline in seconds.

 required

Returns:

Type Description
dict

Record with page name, digest, status, text and latency in seconds.
dict

Failure records include an error description.
Source code in src/inference_lab/client.py 
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
async def infer(  # docvet: ignore[missing-raises]
    client: httpx.AsyncClient, page: Page, timeout: float
) -> dict:
    """Return one terminal record, including failure and observed latency.

    HTTP, timeout and malformed-response failures become records rather than
    escaping to the caller. Task cancellation still propagates.

    Args:
        client: Async HTTP client configured with the server base URL.
        page: Image and expected text prepared before timing.
        timeout: Total request deadline in seconds.

    Returns:
        Record with page name, digest, status, text and latency in seconds.
        Failure records include an error description.
    """
    start = time.perf_counter()
    record = {"page": page.name, "sha256": page.sha256, "status": "error", "text": ""}
    try:
        async with asyncio.timeout(timeout):
            response = await client.post(
                "/v1/chat/completions",
                json={
                    "model": MODEL,
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {"type": "text", "text": PROMPT},
                                {
                                    "type": "image_url",
                                    "image_url": {"url": page.data_url},
                                },
                            ],
                        }
                    ],
                    "max_tokens": 8744,
                    "temperature": 0.0,
                    "repetition_penalty": 1.1,
                    "top_k": 1,
                    "skip_special_tokens": False,
                },
            )
            response.raise_for_status()
            choice = response.json()["choices"][0]
            text = choice["message"]["content"]
            if not isinstance(text, str):
                raise TypeError("response content is not text")  # noqa: TRY301 - record below
            record["text"] = text
            record["finish_reason"] = choice.get("finish_reason")
            missing = [
                s for s in page.expected if normalized(s) not in normalized(text)
            ]
            visible = re.sub(r"<[^>]*>", "", text)
            if choice.get("finish_reason") != "stop":
                record.update(
                    status="incomplete", error="generation did not finish with stop"
                )
            elif not normalized(visible) or missing:
                record.update(
                    status="quality_failed",
                    error="empty output or missing expected text",
                    missing=missing,
                )
            else:
                record["status"] = "success"
    except (TimeoutError, httpx.TimeoutException):
        record.update(status="timeout", error=f"request exceeded {timeout:g}s deadline")
    except httpx.HTTPStatusError as exc:
        record["error"] = f"HTTP {exc.response.status_code}; inspect server logs"
    except httpx.RequestError:
        record["error"] = "connection failed; check inference-axi status"
    except (ValueError, KeyError, IndexError, TypeError):
        record["error"] = "invalid response; expected choices[0].message.content text"
    record["latency_s"] = time.perf_counter() - start
    return record

normalized(text: str) -> str

Casefold text and join word tokens with single spaces.

Parameters:

Name Type Description Default
text str

Expected or extracted text.

 required

Returns:

Type Description
str

Normalized text for substring sanity checks.
Source code in src/inference_lab/client.py 
87
88
89
90
91
92
93
94
95
96
def normalized(text: str) -> str:
    """Casefold text and join word tokens with single spaces.

    Args:
        text: Expected or extracted text.

    Returns:
        Normalized text for substring sanity checks.
    """
    return " ".join(re.findall(r"\w+", text.casefold()))

prepare(path: Path, expected: tuple[str, ...] = ()) -> Page

Decode and normalize before inference timing begins.

Parameters:

Name Type Description Default
path Path

Image file to read and convert to RGB PNG.

 required
expected tuple[str, ...]

Required text snippets for response validation.

 ()

Returns:

Type Description
Page

Prepared image with the original file digest and encoded PNG.
Source code in src/inference_lab/client.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def prepare(path: Path, expected: tuple[str, ...] = ()) -> Page:
    """Decode and normalize before inference timing begins.

    Args:
        path: Image file to read and convert to RGB PNG.
        expected: Required text snippets for response validation.

    Returns:
        Prepared image with the original file digest and encoded PNG.
    """
    raw = path.read_bytes()
    with Image.open(io.BytesIO(raw)) as image:
        image.load()
        output = io.BytesIO()
        image.convert("RGB").save(output, format="PNG")
    data = base64.b64encode(output.getvalue()).decode("ascii")
    return Page(
        path.name,
        hashlib.sha256(raw).hexdigest(),
        f"data:image/png;base64,{data}",
        expected,
    )

inference_lab.experiment

Run finite closed-loop sweeps with explicit result accounting.

Examples:

Summarize observed latencies without claiming tail percentiles:

from inference_lab.experiment import distribution

median = distribution([1.0, 2.0, 3.0])["median_s"]  # 2.0 seconds
See Also

inference_lab.client: Request preparation and response validation. inference_lab.corpus: Synthetic inputs with known expected text.

distribution(values: list[float]) -> dict

Summarize small samples with minimum, median and maximum latency.

Parameters:

Name Type Description Default
values list[float]

Observed request latencies in seconds.

 required

Returns:

Type Description
dict

Count and latency statistics; all statistics are None for no samples.
Source code in src/inference_lab/experiment.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def distribution(values: list[float]) -> dict:
    """Summarize small samples with minimum, median and maximum latency.

    Args:
        values: Observed request latencies in seconds.

    Returns:
        Count and latency statistics; all statistics are None for no samples.
    """
    return {
        "count": len(values),
        "min_s": min(values) if values else None,
        "median_s": statistics.median(values) if values else None,
        "max_s": max(values) if values else None,
    }

load_corpus(directory: Path) -> list[Page]

Load and prepare the synthetic corpus described by its manifest.

Parameters:

Name Type Description Default
directory Path

Directory containing manifest.json and its image files.

 required

Returns:

Type Description
list[Page]

Prepared images in manifest order with required text snippets.

Raises:

Type Description
ValueError

An image escapes the corpus directory, expected text is missing or invalid, or the manifest contains no pages.
Source code in src/inference_lab/experiment.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def load_corpus(directory: Path) -> list[Page]:
    """Load and prepare the synthetic corpus described by its manifest.

    Args:
        directory: Directory containing manifest.json and its image files.

    Returns:
        Prepared images in manifest order with required text snippets.

    Raises:
        ValueError: An image escapes the corpus directory, expected text is
            missing or invalid, or the manifest contains no pages.
    """
    manifest = json.loads((directory / "manifest.json").read_text())
    pages = []
    for item in manifest["pages"]:
        path = (directory / item["file"]).resolve()
        if not path.is_relative_to(directory.resolve()):
            raise ValueError("corpus image must stay inside corpus directory")
        expected = item["expected"]
        if not expected or any(
            not isinstance(s, str) or not s.strip() for s in expected
        ):
            raise ValueError("every corpus page needs nonempty expected text")
        pages.append(prepare(path, tuple(expected)))
    if not pages:
        raise ValueError("corpus contains 0 pages")
    return pages

render(report: dict) -> str

Render a sweep report as Markdown with timing limits and metadata.

Parameters:

Name Type Description Default
report dict

Report containing sweeps and metadata; each sweep must have nonempty latency observations.

 required

Returns:

Type Description
str

Markdown summary table and JSON configuration block.
Source code in src/inference_lab/experiment.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def render(report: dict) -> str:
    """Render a sweep report as Markdown with timing limits and metadata.

    Args:
        report: Report containing sweeps and metadata; each sweep must have
            nonempty latency observations.

    Returns:
        Markdown summary table and JSON configuration block.
    """
    lines = [
        "# Nemotron concurrency smoke demo",
        f"Run status: {report.get('status', 'unknown')}",
        "",
        "This finite workload uses a closed loop and a warmed image cache.",
        "It does not establish overload behavior or production latency guarantees.",
        "Tiny synthetic corpus: text sanity checks only, not a general OCR quality benchmark.",
        "Latency starts at HTTP submission and ends after response validation.",
        "It excludes preparation and waiting for a worker.",
        "Raw per-request latencies and failed outputs are in results.json. "
        "Timeouts are censored observations.",
        "",
        "| Concurrency | Inputs | Success | Failures | Timeouts | Elapsed s | Pages/s | "
        "All median s | All min–max s |",  # noqa: RUF001 - typographic range in prose
        "|---|---|---|---|---|---|---|---|---|",
    ]
    for run in report["sweeps"]:
        d = run["all_latency"]
        lines.append(
            f"| {run['concurrency']} | {run['input_count']} | {run['successful_completions']} | "
            f"{run['failures']} | {run['timeouts']} | {run['elapsed_s']:.3f} | "
            f"{run['successful_pages_per_second']:.3f} | {d['median_s']:.3f} | "
            f"{d['min_s']:.3f}–{d['max_s']:.3f} |"  # noqa: RUF001 - typographic range
        )
    lines += [
        "",
        "## Configuration and provenance",
        "",
        "```json",
        json.dumps(report["metadata"], indent=2),
        "```",
        "",
    ]
    return "\n".join(lines)

summarize(records: list[dict], elapsed: float, concurrency: int) -> dict

Count every terminal status and summarize throughput and latency.

Parameters:

Name Type Description Default
records list[dict]

Request records with status and latency_s fields.

 required
elapsed float

Wall-clock sweep duration in seconds.

 required
concurrency int

Worker limit used by the sweep.

 required

Returns:

Type Description
dict

Completion and failure counts, status counts, throughput, and latency
dict

distributions for all requests and successful requests separately.
Source code in src/inference_lab/experiment.py 
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
def summarize(records: list[dict], elapsed: float, concurrency: int) -> dict:
    """Count every terminal status and summarize throughput and latency.

    Args:
        records: Request records with status and latency_s fields.
        elapsed: Wall-clock sweep duration in seconds.
        concurrency: Worker limit used by the sweep.

    Returns:
        Completion and failure counts, status counts, throughput, and latency
        distributions for all requests and successful requests separately.
    """
    counts = Counter(r["status"] for r in records)
    return {
        "concurrency": concurrency,
        "input_count": len(records),
        "successful_completions": counts["success"],
        "failures": len(records) - counts["success"],
        "timeouts": counts["timeout"],
        "status_counts": dict(counts),
        "elapsed_s": elapsed,
        "successful_pages_per_second": counts["success"] / elapsed if elapsed else 0,
        "all_latency": distribution([r["latency_s"] for r in records]),
        "successful_latency": distribution(
            [r["latency_s"] for r in records if r["status"] == "success"]
        ),
    }

sweep(client: httpx.AsyncClient, pages: list[Page], concurrency: int, timeout: float) -> dict async

Submit each prepared page once through a bounded worker pool.

Preparation and warmup belong outside this measured sweep. Worker tasks share one iterator, so a page is assigned to exactly one worker.

Parameters:

Name Type Description Default
client AsyncClient

Async HTTP client configured with the parser base URL.

 required
pages list[Page]

Nonempty sequence of prepared images.

 required
concurrency int

Worker limit, one of 1, 2, 4 or 8.

 required
timeout float

Deadline in seconds for each request.

 required

Returns:

Type Description
dict

Sweep summary and terminal request records in input order.

Raises:

Type Description
ValueError

Pages are empty or concurrency is unsupported.
Source code in src/inference_lab/experiment.py 
 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
async def sweep(
    client: httpx.AsyncClient, pages: list[Page], concurrency: int, timeout: float
) -> dict:
    """Submit each prepared page once through a bounded worker pool.

    Preparation and warmup belong outside this measured sweep. Worker tasks
    share one iterator, so a page is assigned to exactly one worker.

    Args:
        client: Async HTTP client configured with the parser base URL.
        pages: Nonempty sequence of prepared images.
        concurrency: Worker limit, one of 1, 2, 4 or 8.
        timeout: Deadline in seconds for each request.

    Returns:
        Sweep summary and terminal request records in input order.

    Raises:
        ValueError: Pages are empty or concurrency is unsupported.
    """
    if not pages or concurrency not in (1, 2, 4, 8):
        raise ValueError("sweep needs pages and concurrency 1, 2, 4 or 8")
    queue = iter(enumerate(pages))
    records = []

    async def worker():
        for index, page in queue:
            record = await infer(client, page, timeout)
            record["index"] = index
            records.append(record)

    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        for _ in range(min(concurrency, len(pages))):
            group.create_task(worker())
    elapsed = time.perf_counter() - start
    records.sort(key=lambda r: r["index"])
    return {**summarize(records, elapsed, concurrency), "requests": records}

inference_lab.corpus

Generate eight original, deterministic RGB documents with known text.

Examples:

Create a corpus in a chosen output directory:

from pathlib import Path
from inference_lab.corpus import generate

count = generate(Path("corpus"))
See Also

inference_lab.experiment.load_corpus: Read and prepare generated images.

generate(directory: Path) -> int

Write synthetic page images and their expected-text manifest.

Existing generated filenames and the manifest are replaced.

Parameters:

Name Type Description Default
directory Path

Destination directory, created with any missing parents.

 required

Returns:

Type Description
int

Number of generated pages, always eight.
Source code in src/inference_lab/corpus.py 
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
def generate(directory: Path) -> int:
    """Write synthetic page images and their expected-text manifest.

    Existing generated filenames and the manifest are replaced.

    Args:
        directory: Destination directory, created with any missing parents.

    Returns:
        Number of generated pages, always eight.
    """
    directory.mkdir(parents=True, exist_ok=True)
    pages = []
    font = ImageFont.load_default(size=32)
    for index in range(8):
        lines = [
            f"INFERENCE LAB PAGE {index + 1}",
            "Synthetic inventory record",
            f"Item: cobalt notebook {index + 1}",
            f"Quantity: {12 + index}",
            f"Reference: LAB{401 + index}",
            "All entries are fictional.",
        ]
        image = Image.new("RGB", (1000, 1200), "white")
        draw = ImageDraw.Draw(image)
        for row, line in enumerate(lines):
            draw.text((70, 80 + row * 100), line, fill="black", font=font)
        filename = f"page-{index + 1:02}.png"
        image.save(directory / filename)
        pages.append({"file": filename, "expected": lines})
    (directory / "manifest.json").write_text(
        json.dumps(
            {
                "provenance": (
                    "Original synthetic text and layout generated by inference-lab; "
                    "CC0-1.0. Pillow bundled default font."
                ),
                "pages": pages,
            },
            indent=2,
        )
        + "\n"
    )
    return len(pages)

inference_lab.serving

Inspect and manage only this checkout's labelled Podman container.

The profile in serving/config.json supplies the image, GPU memory floor and server arguments. Ownership checks precede container mutations. This module never stops unrelated GPU workloads.

Examples:

Read the checked-in profile without contacting the GPU or Podman:

from inference_lab import serving

profile = serving.config()
print(profile["container"])
See Also

inference_lab.local_gpu: Explicit opt-in service handover and restoration.

command(argv: list[str], timeout: float = 20) -> str

Run a local tool with captured output and a deadline.

Parameters:

Name Type Description Default
argv list[str]

Trusted executable and arguments, passed directly without a shell.

 required
timeout float

Maximum execution time in seconds.

 20

Returns:

Type Description
str

Standard output with leading and trailing whitespace removed.

Raises:

Type Description
RuntimeError

If the tool exits unsuccessfully; includes bounded stderr.
TimeoutExpired

If execution exceeds the deadline.
OSError

If the executable cannot be started.
Source code in src/inference_lab/serving.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def command(argv: list[str], timeout: float = 20) -> str:
    """Run a local tool with captured output and a deadline.

    Args:
        argv: Trusted executable and arguments, passed directly without a shell.
        timeout: Maximum execution time in seconds.

    Returns:
        Standard output with leading and trailing whitespace removed.

    Raises:
        RuntimeError: If the tool exits unsuccessfully; includes bounded stderr.
        subprocess.TimeoutExpired: If execution exceeds the deadline.
        OSError: If the executable cannot be started.
    """
    # Callers supply local tool argument vectors; no shell evaluates their contents.
    result = subprocess.run(  # noqa: S603
        argv, capture_output=True, text=True, timeout=timeout, check=False
    )
    if result.returncode:
        raise RuntimeError(f"{argv[0]} failed: {result.stderr.strip()[:500]}")
    return result.stdout.strip()

config() -> dict

Read the checkout's runtime profile.

Returns:

Type Description
dict

Parsed settings from serving/config.json.

Raises:

Type Description
OSError

If the profile cannot be read.
JSONDecodeError

If the profile is not valid JSON.
Source code in src/inference_lab/serving.py 
33
34
35
36
37
38
39
40
41
42
43
def config() -> dict:
    """Read the checkout's runtime profile.

    Returns:
        Parsed settings from ``serving/config.json``.

    Raises:
        OSError: If the profile cannot be read.
        json.JSONDecodeError: If the profile is not valid JSON.
    """
    return json.loads((ROOT / "serving/config.json").read_text())

gpu() -> dict

Query GPU zero with nvidia-smi.

Returns:

Type Description
dict

String values for name, UUID, driver, memory in MiB and utilization.

Raises:

Type Description
RuntimeError

If nvidia-smi exits unsuccessfully.
ValueError

If the response does not contain the expected fields.
Source code in src/inference_lab/serving.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def gpu() -> dict:
    """Query GPU zero with nvidia-smi.

    Returns:
        String values for name, UUID, driver, memory in MiB and utilization.

    Raises:
        RuntimeError: If nvidia-smi exits unsuccessfully.
        ValueError: If the response does not contain the expected fields.
    """
    fields = "name,uuid,driver_version,memory.total,memory.free,utilization.gpu"
    values = command(
        [
            "nvidia-smi",
            "--id=0",
            f"--query-gpu={fields}",
            "--format=csv,noheader,nounits",
        ]
    ).split(", ")
    return dict(zip(fields.split(","), values, strict=True))

inspect() -> dict | None

Inspect the configured container after verifying checkout ownership.

Returns:

Type Description
dict | None

Podman inspection data, or None when the container is absent.

Raises:

Type Description
RuntimeError

If Podman fails or the name belongs to another checkout.
See Also

stop: Remove only the inspected immutable ID.

Source code in src/inference_lab/serving.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def inspect() -> dict | None:
    """Inspect the configured container after verifying checkout ownership.

    Returns:
        Podman inspection data, or None when the container is absent.

    Raises:
        RuntimeError: If Podman fails or the name belongs to another checkout.

    See Also:
        [stop][inference_lab.serving.stop]: Remove only the inspected immutable ID.
    """
    # A successful list distinguishes absence from Podman failures.
    names = command(["podman", "ps", "-a", "--format", "{{.Names}}"]).splitlines()
    name = config()["container"]
    if name not in names:
        return None
    item = json.loads(command(["podman", "inspect", name]))[0]
    if item["Config"].get("Labels", {}).get(LABEL) != str(ROOT):
        raise RuntimeError(
            "container name belongs to another checkout; refusing to modify it"
        )
    return item

start() -> dict

Start the owned parser container and wait up to 300 seconds for readiness.

An existing running container is reported without a new readiness check. A failed readiness check leaves the container available for log inspection. Keyboard interruption after launch removes the owned container.

Returns:

Type Description
dict

Container name and readiness or already-running status.

Raises:

Type Description
RuntimeError

If ownership, free memory, startup or readiness checks fail.
TimeoutExpired

If a local tool exceeds its deadline.
KeyboardInterrupt

If startup is interrupted; cleanup runs after launch.
See Also

stop: Remove a failed container after inspecting its logs.

Source code in src/inference_lab/serving.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def start() -> dict:
    """Start the owned parser container and wait up to 300 seconds for readiness.

    An existing running container is reported without a new readiness check.
    A failed readiness check leaves the container available for log inspection.
    Keyboard interruption after launch removes the owned container.

    Returns:
        Container name and readiness or already-running status.

    Raises:
        RuntimeError: If ownership, free memory, startup or readiness checks fail.
        subprocess.TimeoutExpired: If a local tool exceeds its deadline.
        KeyboardInterrupt: If startup is interrupted; cleanup runs after launch.

    See Also:
        [stop][inference_lab.serving.stop]: Remove a failed container after inspecting its logs.
    """
    settings = config()
    item = inspect()
    if item:
        if item["State"]["Running"]:
            return {
                "container": settings["container"],
                "status": "already running; use status to check readiness",
            }
        raise RuntimeError(
            "demo container is stopped; run inference-axi stop before starting again"
        )
    hardware = gpu()
    if int(hardware["memory.free"]) < settings["minimum_free_mib"]:
        raise RuntimeError(
            f"GPU has {hardware['memory.free']} MiB free; this profile requires "
            f"{settings['minimum_free_mib']} MiB. Existing workloads are untouched."
        )
    argv = [
        "podman",
        "run",
        "-d",
        "--name",
        settings["container"],
        "--label",
        f"{LABEL}={ROOT}",
        "--device",
        "nvidia.com/gpu=0",
        "--security-opt",
        "label=disable",
        "--shm-size",
        "2g",
        "-p",
        f"127.0.0.1:{settings['port']}:8000",
    ]
    for key, value in settings["environment"].items():
        argv += ["-e", f"{key}={value}"]
    argv += [settings["image"], *settings["arguments"]]
    command(argv, timeout=60)
    try:
        deadline = time.monotonic() + 300
        with httpx.Client(timeout=3) as client:
            while time.monotonic() < deadline:
                item = inspect()
                if not item or not item["State"]["Running"]:
                    raise RuntimeError("server exited; run inference-axi logs")
                try:
                    response = client.get(f"http://127.0.0.1:{settings['port']}/health")
                    if response.status_code == HTTPStatus.OK:
                        return {"container": settings["container"], "status": "ready"}
                except httpx.RequestError:
                    pass
                time.sleep(2)
        raise RuntimeError(
            "readiness deadline exceeded; run inference-axi logs then inference-axi stop"
        )
    except KeyboardInterrupt:
        stop()
        raise

status() -> dict

Collect container state and GPU information.

Returns:

Type Description
dict

Container state, including absent, and the GPU query result.

Raises:

Type Description
RuntimeError

If ownership verification or a local tool fails.
Source code in src/inference_lab/serving.py 
117
118
119
120
121
122
123
124
125
126
127
def status() -> dict:
    """Collect container state and GPU information.

    Returns:
        Container state, including ``absent``, and the GPU query result.

    Raises:
        RuntimeError: If ownership verification or a local tool fails.
    """
    item = inspect()
    return {"container": item["State"]["Status"] if item else "absent", "gpu": gpu()}

stop() -> dict

Remove the owned parser container by its inspected immutable ID.

Returns:

Type Description
dict

Removal status, or already absent when no container exists.

Raises:

Type Description
RuntimeError

If ownership verification or removal fails.
Source code in src/inference_lab/serving.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def stop() -> dict:
    """Remove the owned parser container by its inspected immutable ID.

    Returns:
        Removal status, or ``already absent`` when no container exists.

    Raises:
        RuntimeError: If ownership verification or removal fails.
    """
    item = inspect()
    if not item:
        return {"status": "already absent"}
    # Use the inspected immutable ID, never a broad name filter.
    command(["podman", "rm", "-f", item["Id"]], timeout=40)
    return {"status": "stopped and removed demo container"}

inference_lab.local_gpu

Provide opt-in GPU service handover with durable restoration evidence.

The context manager preserves the prior service state and loaded model IDs. Callers must explicitly enable handover before this module changes services. Disabled handover leaves the caller's server lifecycle unchanged.

Examples:

Disable handover when using an already managed parser server:

from pathlib import Path
from inference_lab.local_gpu import parser_session

with parser_session(Path("runs/example"), enabled=False):
    pass
See Also

inference_lab.serving: Container ownership and free-memory safeguards.

loaded_models(client: httpx.Client) -> list[str]

Read loaded model IDs from a router in a stable state.

Parameters:

Name Type Description Default
client Client

HTTP client configured with the router's base URL.

 required

Returns:

Type Description
list[str]

IDs whose router state is loaded.

Raises:

Type Description
RuntimeError

If any model is neither loaded nor unloaded.
HTTPError

If the router request fails.
KeyError

If the response lacks required model fields.
Source code in src/inference_lab/local_gpu.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def loaded_models(client: httpx.Client) -> list[str]:
    """Read loaded model IDs from a router in a stable state.

    Args:
        client: HTTP client configured with the router's base URL.

    Returns:
        IDs whose router state is ``loaded``.

    Raises:
        RuntimeError: If any model is neither loaded nor unloaded.
        httpx.HTTPError: If the router request fails.
        KeyError: If the response lacks required model fields.
    """
    response = client.get("/models")
    response.raise_for_status()
    models = response.json()["data"]
    if any(x["status"]["value"] not in ("loaded", "unloaded") for x in models):
        raise RuntimeError("Router model state is changing; retry after it settles")
    return [x["id"] for x in models if x["status"]["value"] == "loaded"]

parser_session(output: Path, enabled: bool) -> Generator[None]

Temporarily run the parser and restore the previous user service state.

Disabled sessions yield without inspecting or modifying services. Enabled sessions record initial state before stopping anything. Cleanup attempts parser removal and service restoration even if the experiment fails. Restoration failures are recorded and raised, including during unwinding.

Parameters:

Name Type Description Default
output Path

Existing run directory for a new lifecycle.json audit file.

 required
enabled bool

Explicit authorization to perform the local service handover.

 required

Yields:

Type Description
Generator[None]

None while the caller runs its experiment.

Raises:

Type Description
RuntimeError

If ownership, service state, startup or restoration fails.
OSError

If the audit file or local commands cannot be accessed.
HTTPError

If the initial model-state request fails.
TimeoutExpired

If a local lifecycle command times out.
See Also

restore_models: Verify model identities. inference_lab.serving.start: Start only the owned parser container.

Source code in src/inference_lab/local_gpu.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@contextmanager
def parser_session(output: Path, enabled: bool) -> Generator[None]:
    """Temporarily run the parser and restore the previous user service state.

    Disabled sessions yield without inspecting or modifying services. Enabled
    sessions record initial state before stopping anything. Cleanup attempts
    parser removal and service restoration even if the experiment fails.
    Restoration failures are recorded and raised, including during unwinding.

    Args:
        output: Existing run directory for a new ``lifecycle.json`` audit file.
        enabled: Explicit authorization to perform the local service handover.

    Yields:
        None while the caller runs its experiment.

    Raises:
        RuntimeError: If ownership, service state, startup or restoration fails.
        OSError: If the audit file or local commands cannot be accessed.
        httpx.HTTPError: If the initial model-state request fails.
        subprocess.TimeoutExpired: If a local lifecycle command times out.

    See Also:
        [restore_models][inference_lab.local_gpu.restore_models]: Verify model identities.
        [inference_lab.serving.start][]: Start only the owned parser container.
    """
    if not enabled:
        yield
        return
    if serving.inspect() is not None:
        raise RuntimeError(
            "An owned parser container already exists; stop it before handover"
        )
    # Fixed arguments query the user service; PATH selects the installed systemctl.
    state = subprocess.run(  # noqa: S603
        ["systemctl", "--user", "is-active", "--quiet", SERVICE],  # noqa: S607
        check=False,
    )
    if state.returncode not in (0, 3):
        raise RuntimeError("Cannot determine the existing llama.cpp service state")
    active = state.returncode == 0
    audit = {
        "service_was_active": active,
        "previous_models": [],
        "experiment_error": None,
        "restoration_errors": [],
        "restoration_verified": False,
    }
    with httpx.Client(base_url=ROUTER, timeout=10) as client:
        if active:
            audit["previous_models"] = loaded_models(client)
        path = output / "lifecycle.json"
        with path.open("x") as file:
            json.dump(audit, file, indent=2)
        try:
            if active:
                service("stop")
            serving.start()
            yield
        except BaseException as exc:
            audit["experiment_error"] = f"{type(exc).__name__}: {exc}"
            raise
        finally:
            try:
                serving.stop()
            except Exception as exc:
                audit["restoration_errors"].append(f"Parser cleanup: {exc}")
            try:
                if active:
                    service("start")
                    restore_models(client, audit["previous_models"])
                else:
                    service("stop")
                audit["restoration_verified"] = not audit["restoration_errors"]
            except Exception as exc:
                audit["restoration_errors"].append(f"Llama restoration: {exc}")
            path.write_text(json.dumps(audit, indent=2))
            if audit["restoration_errors"]:
                raise RuntimeError(f"Restoration failed; see {path}")

restore_models(client: httpx.Client, models: list[str], timeout: float = 180) -> None

Wait for router health, load previous models and verify their identities.

Parameters:

Name Type Description Default
client Client

HTTP client configured with the router's base URL.

 required
models list[str]

Previously loaded model IDs to restore.

 required
timeout float

Shared deadline in seconds for health, loading and verification.

 180

Raises:

Type Description
RuntimeError

If readiness or model restoration exceeds the deadline.
HTTPError

If loading or model verification requests fail.
KeyError

If the router response lacks required model fields.
Source code in src/inference_lab/local_gpu.py 
 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
def restore_models(
    client: httpx.Client, models: list[str], timeout: float = 180
) -> None:
    """Wait for router health, load previous models and verify their identities.

    Args:
        client: HTTP client configured with the router's base URL.
        models: Previously loaded model IDs to restore.
        timeout: Shared deadline in seconds for health, loading and verification.

    Raises:
        RuntimeError: If readiness or model restoration exceeds the deadline.
        httpx.HTTPError: If loading or model verification requests fail.
        KeyError: If the router response lacks required model fields.
    """
    deadline = time.monotonic() + timeout
    while True:
        try:
            response = client.get("/health", timeout=2)
            response.raise_for_status()
            break
        except httpx.HTTPError:
            if time.monotonic() >= deadline:
                raise RuntimeError("Router readiness deadline exceeded") from None
            time.sleep(1)
    for model in models:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise RuntimeError("Model restore deadline exceeded")
        response = client.post("/models/load", json={"model": model}, timeout=remaining)
        response.raise_for_status()
    while time.monotonic() < deadline:
        response = client.get(
            "/models", timeout=min(5, max(0.1, deadline - time.monotonic()))
        )
        response.raise_for_status()
        actual = {
            x["id"] for x in response.json()["data"] if x["status"]["value"] == "loaded"
        }
        if actual == set(models):
            return
        time.sleep(1)
    raise RuntimeError("Previous model IDs did not reach loaded state before deadline")

service(action: str) -> None

Apply an action to the configured user service.

Parameters:

Name Type Description Default
action str

systemctl action, such as start or stop.

 required

Raises:

Type Description
RuntimeError

If systemctl exits unsuccessfully.
TimeoutExpired

If the command exceeds 60 seconds.
Source code in src/inference_lab/local_gpu.py 
37
38
39
40
41
42
43
44
45
46
47
def service(action: str) -> None:
    """Apply an action to the configured user service.

    Args:
        action: systemctl action, such as ``start`` or ``stop``.

    Raises:
        RuntimeError: If systemctl exits unsuccessfully.
        subprocess.TimeoutExpired: If the command exceeds 60 seconds.
    """
    serving.command(["systemctl", "--user", action, SERVICE], timeout=60)

inference_lab.cli

Run parsing experiments with compact status views and JSON evidence.

Examples:

Inspect the available commands without starting a server:

from inference_lab.cli import parser

print(parser().format_help())
See Also

Parser

Bases: ArgumentParser

Report argument errors in the same flat format as command results.

Examples:

Parse options without contacting hardware:

from inference_lab.cli import parser

options = parser().parse_args(["logs", "--full"])
assert options.full
Source code in src/inference_lab/cli.py 
 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
class Parser(argparse.ArgumentParser):
    """Report argument errors in the same flat format as command results.

    Examples:
        Parse options without contacting hardware:

        ```python
        from inference_lab.cli import parser

        options = parser().parse_args(["logs", "--full"])
        assert options.full
        ```
    """

    def error(self, message: str):
        """Print argument guidance and exit with usage status two.

        Args:
            message: The argument parser's diagnostic.

        Raises:
            SystemExit: Always, with exit status two.
        """
        flags = ", ".join(s for a in self._actions for s in a.option_strings)
        emit({"error": message, "valid_flags": flags, "help": f"{self.prog} --help"})
        raise SystemExit(2)

error(message: str)

Print argument guidance and exit with usage status two.

Parameters:

Name Type Description Default
message str

The argument parser's diagnostic.

 required

Raises:

Type Description
SystemExit

Always, with exit status two.
Source code in src/inference_lab/cli.py 
100
101
102
103
104
105
106
107
108
109
110
111
def error(self, message: str):
    """Print argument guidance and exit with usage status two.

    Args:
        message: The argument parser's diagnostic.

    Raises:
        SystemExit: Always, with exit status two.
    """
    flags = ", ".join(s for a in self._actions for s in a.option_strings)
    emit({"error": message, "valid_flags": flags, "help": f"{self.prog} --help"})
    raise SystemExit(2)

emit(data: dict) -> None

Print a flat status mapping to standard output.

Parameters:

Name Type Description Default
data dict

Ordered keys and scalar values to display.

 required
Source code in src/inference_lab/cli.py 
75
76
77
78
79
80
81
82
83
def emit(data: dict) -> None:
    """Print a flat status mapping to standard output.

    Args:
        data: Ordered keys and scalar values to display.
    """
    # Deliberately limited to this CLI's flat scalar views, not a generic codec.
    for key, value in data.items():
        print(f"{key}: {scalar(value)}")

main() -> int

Dispatch a command and report recoverable failures as status views.

Returns:

Type Description
int

Zero on success, one on command failure, or 130 on interruption.

Raises:

Type Description
SystemExit

Argument parsing requests help or rejects usage.
Source code in src/inference_lab/cli.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def main() -> int:
    """Dispatch a command and report recoverable failures as status views.

    Returns:
        Zero on success, one on command failure, or 130 on interruption.

    Raises:
        SystemExit: Argument parsing requests help or rejects usage.
    """
    args = parser().parse_args()
    try:
        if args.action in (None, "status"):
            _show_status()
        elif args.action == "corpus":
            emit(
                {
                    "pages": generate(args.output),
                    "directory": str(args.output),
                    "help": "inference-axi bench <corpus-directory> --output <new-run-directory>",
                }
            )
        elif args.action in ("start", "stop"):
            emit(getattr(serving, args.action)())
        elif args.action == "build":
            _build_image()
        elif args.action == "logs":
            _show_logs(args.full)
        elif args.action == "ocr":
            return asyncio.run(run_ocr(args))
        elif args.action == "bench":
            return asyncio.run(run_bench(args))
    except KeyboardInterrupt:
        emit(
            {
                "error": "interrupted; completed sweep files are retained",
                "help": "inference-axi stop",
            }
        )
        return 130
    except (
        OSError,
        RuntimeError,
        ValueError,
        KeyError,
        subprocess.SubprocessError,
    ) as exc:
        emit(
            {
                "error": str(exc),
                "help": "inference-axi --help; see docs/reference/evidence.md",
            }
        )
        return 1
    return 0

parser() -> Parser

Build the argument parser without probing the host or GPU.

Returns:

Type Description
Parser

The parser with all supported subcommands.
Source code in src/inference_lab/cli.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def parser() -> Parser:
    """Build the argument parser without probing the host or GPU.

    Returns:
        The parser with all supported subcommands.
    """
    root = Parser(
        prog="inference-axi",
        description="Local Nemotron OCR smoke demo; no arguments shows live state.",
        allow_abbrev=False,
    )
    sub = root.add_subparsers(dest="action")
    for name in ("status", "start", "stop", "build"):
        sub.add_parser(name, allow_abbrev=False)
    logs = sub.add_parser("logs", allow_abbrev=False)
    logs.add_argument("--full", action="store_true")
    corpus = sub.add_parser("corpus", allow_abbrev=False)
    corpus.add_argument("--output", type=Path, default=Path("corpus"))
    for name in ("ocr", "bench"):
        p = sub.add_parser(name, allow_abbrev=False)
        p.add_argument("input", type=Path)
        p.add_argument("--output", required=True, type=Path)
        p.add_argument("--timeout", type=positive, default=120.0)
        p.add_argument("--url", default="http://127.0.0.1:8017")
        if name == "ocr":
            p.add_argument("--expect", action="append", default=[])
            p.add_argument("--full", action="store_true")
        else:
            p.add_argument(
                "--concurrency",
                nargs="+",
                type=int,
                choices=(1, 2, 4, 8),
                default=[1, 2, 4, 8],
            )
    return root

positive(value: str) -> float

Parse a finite, strictly positive timeout.

Parameters:

Name Type Description Default
value str

Numeric command-line argument.

 required

Returns:

Type Description
float

The parsed timeout in seconds.

Raises:

Type Description
ValueError

The input is not numeric.
ArgumentTypeError

The value is nonpositive or nonfinite.
Source code in src/inference_lab/cli.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def positive(value: str) -> float:
    """Parse a finite, strictly positive timeout.

    Args:
        value: Numeric command-line argument.

    Returns:
        The parsed timeout in seconds.

    Raises:
        ValueError: The input is not numeric.
        argparse.ArgumentTypeError: The value is nonpositive or nonfinite.
    """
    result = float(value)
    if not math.isfinite(result) or result <= 0:
        raise argparse.ArgumentTypeError("must be a finite positive number")
    return result

run_bench(args: argparse.Namespace) -> int async

Warm the corpus and record bounded concurrency sweeps.

Parameters:

Name Type Description Default
args Namespace

Parsed benchmark options and a new output directory.

 required

Returns:

Type Description
int

Zero if every measured request succeeds, otherwise one.

Raises:

Type Description
RuntimeError

The owned server is unavailable or a warmup fails.
ValueError

The URL differs from the managed server.
BaseException

Run failures propagate after partial evidence is saved.
Source code in src/inference_lab/cli.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
async def run_bench(args: argparse.Namespace) -> int:
    """Warm the corpus and record bounded concurrency sweeps.

    Args:
        args: Parsed benchmark options and a new output directory.

    Returns:
        Zero if every measured request succeeds, otherwise one.

    Raises:
        RuntimeError: The owned server is unavailable or a warmup fails.
        ValueError: The URL differs from the managed server.
        BaseException: Run failures propagate after partial evidence is saved.
    """
    preparation_start = time.perf_counter()
    pages = load_corpus(args.input)
    preparation_s = time.perf_counter() - preparation_start
    # Require our running container so intended config cannot masquerade as observed config.
    item = serving.inspect()
    if not item or not item["State"]["Running"]:
        raise RuntimeError(
            "benchmark requires this checkout's running server; run inference-axi start"
        )
    if args.url.rstrip("/") != "http://127.0.0.1:8017":
        raise ValueError(
            "benchmark uses the managed local server at http://127.0.0.1:8017"
        )
    args.output.mkdir(parents=True, exist_ok=False)
    metadata = {
        "started_at": datetime.now(UTC).isoformat(),
        "python": platform.python_version(),
        "platform": platform.platform(),
        "gpu_before": serving.gpu(),
        "serving": serving.config(),
        "container_id": item["Id"],
        "image_id": item["Image"],
        "container_args": item["Args"],
        "packages": serving.command(
            ["podman", "exec", item["Id"], "cat", "/serving-packages.txt"]
        ),
        "git_commit": serving.command(
            ["git", "-C", str(serving.ROOT), "rev-parse", "HEAD"]
        ),
        "git_dirty": bool(
            serving.command(["git", "-C", str(serving.ROOT), "status", "--porcelain"])
        ),
        "client_url": args.url,
        "timeout_s": args.timeout,
        "preparation_s": preparation_s,
        "warmup_policy": "one serial request per corpus page before measurement",
        "cache_policy": "prefix cache disabled; default multimodal cache retained and warmed",
        "quality": "all expected phrases after case/punctuation normalization",
        "corpus": [
            {"file": p.name, "sha256": p.sha256, "expected": p.expected} for p in pages
        ],
    }
    report = {"status": "running", "metadata": metadata, "warmup": [], "sweeps": []}

    def save():
        (args.output / "results.json").write_text(json.dumps(report, indent=2) + "\n")
        (args.output / "summary.md").write_text(render(report))

    try:
        async with httpx.AsyncClient(
            base_url=args.url,
            timeout=args.timeout,
            limits=httpx.Limits(max_connections=8),
        ) as client:
            for page in pages:
                record = await infer(client, page, args.timeout)
                report["warmup"].append(record)
                save()
                if record["status"] != "success":
                    raise RuntimeError(  # noqa: TRY301 - outer handler records aborted evidence
                        f"warmup failed; inspect {args.output}/results.json"
                    )
            for concurrency in args.concurrency:
                report["sweeps"].append(
                    await sweep(client, pages, concurrency, args.timeout)
                )
                save()
        report["status"] = "complete"
    except BaseException:
        report["status"] = "aborted"
        raise
    finally:
        metadata["finished_at"] = datetime.now(UTC).isoformat()
        save()
    emit(
        {
            "sweeps": len(report["sweeps"]),
            "failures": sum(s["failures"] for s in report["sweeps"]),
            "summary": str(args.output / "summary.md"),
            "results": str(args.output / "results.json"),
        }
    )
    return 1 if any(s["failures"] for s in report["sweeps"]) else 0

run_ocr(args: argparse.Namespace) -> int async

Extract one image and save its text and terminal request record.

Parameters:

Name Type Description Default
args Namespace

Parsed OCR options, including input, output, URL and timeout.

 required

Returns:

Type Description
int

Zero for a successful request, otherwise one.
Source code in src/inference_lab/cli.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def run_ocr(args: argparse.Namespace) -> int:
    """Extract one image and save its text and terminal request record.

    Args:
        args: Parsed OCR options, including input, output, URL and timeout.

    Returns:
        Zero for a successful request, otherwise one.
    """
    page = prepare(args.input, tuple(args.expect))
    async with httpx.AsyncClient(base_url=args.url, timeout=args.timeout) as client:
        record = await infer(client, page, args.timeout)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(record["text"])
    args.output.with_suffix(args.output.suffix + ".json").write_text(
        json.dumps(record, indent=2) + "\n"
    )
    text = record["text"]
    emit(
        {
            "status": record["status"],
            "output": str(args.output),
            "characters": len(text),
            "preview": text if args.full else text[:PREVIEW_CHARACTERS],
            "quality": "expected-text sanity check"
            if args.expect
            else "nonempty output only; supply --expect for known text",
        }
    )
    if len(text) > PREVIEW_CHARACTERS and not args.full:
        emit(
            {
                "help": (
                    f"Read {args.output} for complete output; "
                    "--full prints all text on a new request"
                )
            }
        )
    if "error" in record:
        emit({"error": record["error"]})
    return 0 if record["status"] == "success" else 1

scalar(value: str | int | float | bool | None) -> str

Encode one scalar for the CLI's flat TOON views.

Parameters:

Name Type Description Default
value str | int | float | bool | None

A string or JSON-compatible scalar.

 required

Returns:

Type Description
str

A quoted or bare scalar with control characters escaped.
Source code in src/inference_lab/cli.py 
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
def scalar(value: str | int | float | bool | None) -> str:
    """Encode one scalar for the CLI's flat TOON views.

    Args:
        value: A string or JSON-compatible scalar.

    Returns:
        A quoted or bare scalar with control characters escaped.
    """
    if not isinstance(value, str):
        return json.dumps(value, allow_nan=False)
    if (
        not value
        or value != value.strip(" \t")
        or value in ("true", "false", "null")
        or re.fullmatch(r"[+-]?[0-9]+(?:\.[0-9]+)?(?:e[+-]?[0-9]+)?", value, re.I)
        or re.search(r'[:,"\\\[\]{}\x00-\x1f]', value)
        or value.startswith(("-", "#"))
    ):
        escapes = {"\\": "\\\\", '"': '\\"', "\n": "\\n", "\r": "\\r", "\t": "\\t"}
        return (
            '"'
            + "".join(
                escapes.get(
                    c, f"\\u{ord(c):04x}" if ord(c) < CONTROL_CHARACTER_LIMIT else c
                )
                for c in value
            )
            + '"'
        )
    return value

inference_lab

Provide the local Nemotron OCR demonstration and receipt diagnostics.

Attributes:

Name Type Description
client

Image preparation and bounded inference requests.
corpus

Synthetic document generation.
experiment

Closed-loop sweeps and result summaries.
receipt_metrics

Diagnostic text and numeric-string comparison.

Examples:

Compare text without starting a GPU server:

from inference_lab.receipt_metrics import score

recall = score("Total 12.00", "Total 12.00")["numeric_recall"]  # 1.0
See Also

inference_lab.cli: Command-line entry point. inference_lab.serving: Owned-container lifecycle and resource checks.