15  Monitor runs

15.1 Save some results to monitor

A pipeline often writes one small result file per image. The block below writes ten such files with two synthetic measurements each. The first is the image’s noise in DN (data numbers, the detector’s count unit). The second is its zero-point ratio. This ratio is the image’s flux scale relative to a reference image, computed from their zero points (the calibration that converts counts to brightness). Image 6 has a bright sky, image 3 has no zero point, and one file is corrupt.

import json
import shutil
from pathlib import Path

import numpy as np

shutil.rmtree("runs/monitor", ignore_errors=True)  # a new monitor needs an empty directory
results = Path("runs/monitor/results")
results.mkdir(parents=True)
rng = np.random.default_rng(4)
start = np.sort(rng.uniform(0, 8, 10))  # hours since the start of the night
noise = rng.normal(12.0, 0.3, 10)
noise[6] = 19.0  # a bright sky
ratio = rng.normal(1.0, 0.01, 10)
for i in range(10):
    result = {"image": f"image-{i:02d}", "start_h": start[i], "exposure_h": 0.1,
              "noise_dn": noise[i], "zero_point_ratio": None if i == 3 else ratio[i]}
    (results / f"image-{i:02d}.json").write_text(json.dumps(result))
(results / "image-10.json").write_text('{"image": "image-10", "noi')  # a truncated file
print(sorted(path.name for path in results.iterdir())[-3:])

['image-08.json', 'image-09.json', 'image-10.json']

15.2 Turn one result into a sample

Write an analyzer, a function that takes one input and returns a Sample. The sample holds the subject’s measurements, the findings of your rules, and references to its files. sequence sets the display order. The observation times use your own unit and origin.

from cuphoton.core.artifacts import file_sha256
from cuphoton.xlens import Artifact, Finding, Measurement, Sample


def analyze(path):
    result = json.loads(path.read_text())
    findings = []
    if result["noise_dn"] > 15:
        findings.append(Finding("high_noise", "Noise is above 15 DN.", category="data_quality"))
    if result["zero_point_ratio"] is None:
        findings.append(Finding("no_zero_point", "Calibration gave no zero point.",
                                category="processing", severity="error"))
    return Sample(
        id=result["image"], sequence=int(result["image"][-2:]),
        observed_start=result["start_h"], observed_end=result["start_h"] + result["exposure_h"], time_unit="h",
        measurements={"noise": Measurement(result["noise_dn"], "DN"),
                      "zero_point_ratio": Measurement(result["zero_point_ratio"])},
        findings=tuple(findings),
        artifacts=(Artifact("result", f"results/{path.name}", file_sha256(path), "application/json"),),
    )


print(analyze(results / "image-03.json").measurements["zero_point_ratio"])

Measurement(value=None, unit='', status='missing', coverage=None, note='')

A Measurement holds one scalar with its unit and status. A None value gets the status missing, and NaN becomes invalid. A Finding is the result of one rule. Its category (scientific, data_quality or processing) is separate from its severity (info, warning or error). An Artifact names a file by relative path and SHA-256. xLens never opens it. Keep arrays and images in such files.

15.3 Replay the results into a monitor

A Monitor owns one state directory and rewrites state.json there after every change. replay passes each input to the analyzer in order and publishes the returned samples. When the analyzer raises, xLens records the error and goes on with the next input.

from cuphoton.xlens import Monitor

monitor = Monitor("runs/monitor/state", run_id="night-1")
monitor.replay(sorted(results.glob("*.json")), analyze)
state = monitor.snapshot()
print(len(monitor.samples), "samples, collector", state["collector"])
print("errors:", [(error["source_id"], error["error_type"]) for error in state["errors"]])
print("flagged:", {sample.id: [f.code for f in sample.findings] for sample in monitor.samples if sample.findings})

10 samples, collector complete
errors: [('input-10', 'JSONDecodeError')]
flagged: {'image-03': ['no_zero_point'], 'image-06': ['high_noise']}

The corrupt file became an ingestion error, not a sample. The error record keeps the exception class and a source ID, here the input’s position input-10, but not the message. To ingest one input at a time, for example as files arrive, call monitor.ingest(item, analyze, source_id=...) with a source ID you choose.

15.4 Look at the measurements

Figure 15.1 plots the samples the way the dashboard’s measurement history does.

Plotting code

import matplotlib.pyplot as plt

samples = monitor.samples
hours = np.array([sample.observed_start for sample in samples])
flagged = np.array([bool(sample.findings) for sample in samples])
fig, axes = plt.subplots(2, 1, figsize=(6.5, 3.6), sharex=True)
for ax, name in zip(axes, ["noise", "zero_point_ratio"]):
    values = np.array([np.nan if s.measurements[name].value is None else s.measurements[name].value for s in samples])
    ax.plot(hours, values, "o-", color="#2A78D6", ms=4, lw=1)
    ax.scatter(hours[flagged], values[flagged], s=120, facecolors="none", edgecolors="#EB6834", linewidths=1.5)
    unit = samples[0].measurements[name].unit
    ax.set_ylabel(name + (f" [{unit}]" if unit else ""), fontsize=8)
axes[-1].set_xlabel("Observation start [h]")
fig.text(1, 1, "Synthetic data, for demonstration only", ha="right", va="bottom",
         fontsize=7, style="italic", color="0.4", transform=fig.transFigure)
fig.tight_layout()
plt.show()

15.5 Revise a sample

When you reprocess an image, publish the new evidence under the same id with a higher revision. xLens keeps the old revision in the history. Publishing the same evidence again changes nothing and returns False. Changed evidence with the same revision raises an error.

from dataclasses import replace

old = next(sample for sample in monitor.samples if sample.id == "image-03")
new = replace(old, revision=2, findings=(),
              measurements={**old.measurements, "zero_point_ratio": Measurement(0.998)})
print(monitor.publish(new), monitor.publish(new))
try:
    monitor.publish(replace(new, findings=old.findings))  # changed, same revision
except ValueError as error:
    print("ValueError:", error)
print(len(monitor.samples), "samples,", len(monitor.snapshot()["history"]), "revisions")

True False
ValueError: changed evidence requires an increasing revision
10 samples, 11 revisions

Recompute any totals you keep from monitor.samples, which always holds the current revisions.

15.6 Open a saved run

The state directory is the saved run. Monitor.load opens it again. With recover=False, loading reads without writing, for example while the run’s own process still writes. Default recovery marks unfinished collection as interrupted. Use it only when you own the writer role: loading does not acquire an exclusive lock. Keep one writer per state directory.

viewer = Monitor.load("runs/monitor/state", recover=False)
image_03 = viewer.samples[3]
print(viewer.snapshot()["run_id"], len(viewer.samples), "samples")
print(image_03.id, "revision", image_03.revision, image_03.measurements["zero_point_ratio"].value)

night-1 10 samples
image-03 revision 2 0.998

15.7 Export or serve the dashboard

Dashboard.export writes the run into one HTML file that opens in a browser without a server. The file is a fixed snapshot, and its content security policy blocks network requests.

from cuphoton.xlens import Dashboard

page = Dashboard(viewer, title="Night 1 images").export("runs/monitor/report.html")
print(page, f"{page.stat().st_size / 1024:.0f} KiB")

runs/monitor/report.html 49 KiB

To follow a run while it is written, serve the state directory instead, then open http://127.0.0.1:8765:

cuphoton xlens serve --state runs/monitor/state --port 8765

The server rereads the saved state on every request and stops with Ctrl+C. It answers only the address it listens on, which includes localhost for the default loopback address, and the names you add with --allowed-host. Besides the page, it serves JSON for scripts:

make_server builds the same server in Python. The block below starts it on a free port, reads /health, and stops it:

import threading
import urllib.request

from cuphoton.xlens.dashboard import make_server

with make_server(Dashboard(viewer), host="127.0.0.1", port=0) as server:  # port 0 picks a free port
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{server.server_port}/health") as response:
            print(response.read().decode())
    finally:
        server.shutdown()
        thread.join()

{"collector": "complete", "storage_error": false}

In a pipeline that writes results as it runs, let a MonitorSession poll for new inputs in the process that owns the monitor, and serve the dashboard from there. This poller processes each filename once. Have the producer write a temporary file and rename it to .json only when complete, then leave that file unchanged. Handle reprocessing separately by publishing a higher sample revision.

from cuphoton.xlens import MonitorSession, serve

seen = set()


def poll():  # called every poll_interval seconds
    for path in sorted(results.glob("*.json")):
        if path.name not in seen:
            seen.add(path.name)
            monitor.ingest(path, analyze, source_id=path.name)


serve(Dashboard(monitor, title="Night 1 images"), session=MonitorSession(monitor, poll=poll, poll_interval=10.0))

15.8 What the dashboard shows

The page has these parts:

• Run status: the run ID, the sample count, the collector status (idle, running, complete, stopped, failed or interrupted) and the number of revisions.
• Samples: the list in sequence order, with a text filter and a finding-category filter.
• Measurement history: up to six measurements at a time, against sample sequence or observation time, on a linear or log scale. Samples with findings are outlined. Missing and invalid values leave gaps.
• Selected sample: its findings, a table of measurements with value, unit, status and coverage, your domain view, and its configuration, provenance and artifacts.
• Investigate this run: questions to a language model. They need a live session with an inference configuration.
• Ingestion diagnostics: the errors that ingest and replay recorded.

The page shows the last 256 samples in sequence order. state.json keeps them all.

15.9 Options

A sample is at most 64 KiB with at most 128 measurements. A run holds at most 10,000 samples, 50,000 revisions and 1,000 reports, and keeps the latest 100 ingestion errors. Start another run when you reach a limit.

15.10 Command line

cuphoton xlens quickstart --output-dir xlens-demo                   # an invented run and its report.html
cuphoton xlens export --state runs/monitor/state --output-file runs/monitor/report-cli.html
cuphoton xlens serve --state runs/monitor/state --port 8765
# Backend for a reverse proxy on this machine
cuphoton xlens serve --state runs/monitor/state --host 127.0.0.1 --port 8765 --allowed-host monitor.example

serve takes exactly one of --state and --site. --site serves an installed application’s own collection of monitors. --prefix mounts the page under a URL path.

The server does not configure TLS or authenticate dashboard viewers. When sharing a dashboard, use a proxy that provides both. Bind the backend to 127.0.0.1 for a proxy on the same machine, as shown above. If the proxy runs elsewhere, bind a private interface and restrict access to the backend port to that proxy. Set --allowed-host to the HTTP Host value the proxy forwards, including its port when present; this header check does not authenticate the viewer.

15.11 Troubleshooting

15.12 Next steps

• Produce the values to monitor with the end-to-end pipeline: Chapter 9.
• Run long batches on many GPUs: Chapter 14.
• To ask a language model about saved evidence, install the xlens-inference extra and use cuphoton xlens ask. Refer to the xLens reference, which also covers application sites and domain views.