Skip to content

API Reference

luxspec.telemetry.LuxFrame dataclass

A single calibrated or raw Luxnode scan frame.

The class accepts the compact V0 firmware payload {"device": "...", "kind": "raw", "pixels": [...]} and the richer protocol payload used by the Python SDK.

Source code in src/luxspec/telemetry.py 
 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
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@dataclass(frozen=True)
class LuxFrame:
    """A single calibrated or raw Luxnode scan frame.

    The class accepts the compact V0 firmware payload
    ``{"device": "...", "kind": "raw", "pixels": [...]}`` and the richer
    protocol payload used by the Python SDK.
    """

    counts: np.ndarray
    wavelengths_nm: np.ndarray
    measurement_kind: str = "raw"
    schema_version: str = PROTOCOL_VERSION
    device_id: str = "luxnode"
    firmware_version: str | None = None
    module_id: str | None = None
    frame_id: str | None = None
    timestamp_unix_ms: int | None = None
    dark_counts: np.ndarray | None = None
    reference_counts: np.ndarray | None = None
    telemetry: LuxTelemetry = field(default_factory=LuxTelemetry)
    metadata: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        counts = _array(self.counts, "counts")
        wavelengths = _array(self.wavelengths_nm, "wavelengths_nm")
        if wavelengths.size != counts.size:
            raise ValueError("wavelengths_nm and counts must have the same length.")
        object.__setattr__(self, "counts", counts)
        object.__setattr__(self, "wavelengths_nm", wavelengths)
        for attr in ("dark_counts", "reference_counts"):
            values = getattr(self, attr)
            if values is not None:
                arr = _array(values, attr)
                if arr.size != counts.size:
                    raise ValueError(f"{attr} must match counts length.")
                object.__setattr__(self, attr, arr)

    @classmethod
    def from_mapping(cls, payload: Mapping[str, Any]) -> "LuxFrame":
        counts = payload.get("counts", payload.get("pixels"))
        if counts is None:
            raise ValueError("Lux frame payload needs counts or pixels.")
        count_array = _array(counts, "counts")
        wavelengths = payload.get("wavelengths_nm", payload.get("wavelengths"))
        if wavelengths is None:
            wavelengths = WavelengthCalibration.linear(340, 850, pixels=count_array.size).wavelengths(count_array.size)
        telemetry_payload = payload.get("telemetry")
        telemetry = LuxTelemetry.from_mapping(telemetry_payload if isinstance(telemetry_payload, Mapping) else None)
        integration = payload.get("integration_time_ms")
        if telemetry.integration_time_ms is None and integration is not None:
            telemetry = LuxTelemetry.from_mapping({**telemetry.to_dict(), "integration_time_ms": integration})
        return cls(
            counts=count_array,
            wavelengths_nm=np.asarray(wavelengths, dtype=float),
            measurement_kind=str(payload.get("measurement_kind", payload.get("kind", "raw"))),
            schema_version=str(payload.get("schema_version", PROTOCOL_VERSION)),
            device_id=str(payload.get("device_id", payload.get("device", "luxnode"))),
            firmware_version=payload.get("firmware_version"),
            module_id=payload.get("module_id", payload.get("module")),
            frame_id=payload.get("frame_id"),
            timestamp_unix_ms=int(payload["timestamp_unix_ms"]) if payload.get("timestamp_unix_ms") is not None else None,
            dark_counts=np.asarray(payload["dark_counts"], dtype=float) if payload.get("dark_counts") is not None else None,
            reference_counts=np.asarray(payload["reference_counts"], dtype=float)
            if payload.get("reference_counts") is not None
            else None,
            telemetry=telemetry,
            metadata={str(k): v for k, v in payload.get("metadata", {}).items()} if isinstance(payload.get("metadata"), Mapping) else {},
        )

    @classmethod
    def from_json(cls, text: str) -> "LuxFrame":
        return cls.from_mapping(json.loads(text))

    def to_spectrum(self, kind: str | None = None) -> Spectrum:
        return Spectrum(
            self.wavelengths_nm,
            self.counts,
            kind=kind or self.measurement_kind,
            metadata={
                "device_id": self.device_id,
                "firmware_version": self.firmware_version or "",
                "module_id": self.module_id or "",
                **dict(self.metadata),
            },
        )

    def to_dict(self) -> dict[str, Any]:
        data: dict[str, Any] = {
            "schema_version": self.schema_version,
            "device_id": self.device_id,
            "firmware_version": self.firmware_version,
            "module_id": self.module_id,
            "frame_id": self.frame_id,
            "timestamp_unix_ms": self.timestamp_unix_ms,
            "measurement_kind": self.measurement_kind,
            "wavelengths_nm": self.wavelengths_nm.tolist(),
            "counts": self.counts.tolist(),
            "dark_counts": self.dark_counts.tolist() if self.dark_counts is not None else None,
            "reference_counts": self.reference_counts.tolist() if self.reference_counts is not None else None,
            "telemetry": self.telemetry.to_dict(),
            "metadata": dict(self.metadata),
        }
        return {key: value for key, value in data.items() if value not in (None, {}, [])}

    def to_json(self, *, indent: int | None = None) -> str:
        return json.dumps(self.to_dict(), indent=indent)

luxspec.telemetry.LuxTelemetry dataclass

Device-side telemetry that travels with a Luxnode scan frame.

Source code in src/luxspec/telemetry.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass(frozen=True)
class LuxTelemetry:
    """Device-side telemetry that travels with a Luxnode scan frame."""

    board_temperature_c: float | None = None
    detector_temperature_c: float | None = None
    battery_mv: float | None = None
    usb_mv: float | None = None
    rssi_dbm: float | None = None
    integration_time_ms: float | None = None
    gain: float | None = None
    led_current_ma: float | None = None
    exposure_count: int | None = None
    warnings: tuple[str, ...] = ()
    extra: Mapping[str, Any] = field(default_factory=dict)

    @classmethod
    def from_mapping(cls, payload: Mapping[str, Any] | None) -> "LuxTelemetry":
        if not payload:
            return cls()
        warnings = payload.get("warnings", ())
        if isinstance(warnings, str):
            warnings = (warnings,)
        return cls(
            board_temperature_c=_float_or_none(payload.get("board_temperature_c")),
            detector_temperature_c=_float_or_none(payload.get("detector_temperature_c")),
            battery_mv=_float_or_none(payload.get("battery_mv")),
            usb_mv=_float_or_none(payload.get("usb_mv")),
            rssi_dbm=_float_or_none(payload.get("rssi_dbm")),
            integration_time_ms=_float_or_none(payload.get("integration_time_ms")),
            gain=_float_or_none(payload.get("gain")),
            led_current_ma=_float_or_none(payload.get("led_current_ma")),
            exposure_count=int(payload["exposure_count"]) if payload.get("exposure_count") is not None else None,
            warnings=tuple(str(item) for item in warnings),
            extra={str(k): v for k, v in payload.items() if k not in _TELEMETRY_FIELDS},
        )

    def to_dict(self) -> dict[str, Any]:
        data: dict[str, Any] = {
            "board_temperature_c": self.board_temperature_c,
            "detector_temperature_c": self.detector_temperature_c,
            "battery_mv": self.battery_mv,
            "usb_mv": self.usb_mv,
            "rssi_dbm": self.rssi_dbm,
            "integration_time_ms": self.integration_time_ms,
            "gain": self.gain,
            "led_current_ma": self.led_current_ma,
            "exposure_count": self.exposure_count,
            "warnings": list(self.warnings),
        }
        data.update(dict(self.extra))
        return {key: value for key, value in data.items() if value not in (None, [], {})}

luxspec.client.SerialLuxNode

Bases: BaseLuxNodeClient

Read JSON-line frames from Luxnode firmware over a serial port.

Source code in src/luxspec/client.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 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
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
class SerialLuxNode(BaseLuxNodeClient):
    """Read JSON-line frames from Luxnode firmware over a serial port."""

    COMMANDS = {
        "raw": "SCAN",
        "scan": "SCAN",
        "dark": "DARK",
        "white": "WHITE",
        "blank": "WHITE",
        "reference": "WHITE",
    }

    def __init__(
        self,
        port: str,
        *,
        baud: int = 115200,
        timeout: float = 5.0,
        serial_factory: Any | None = None,
        stream_command: str | None = "STREAM START",
        stop_stream_command: str | None = "STREAM STOP",
    ) -> None:
        self.port = port
        self.baud = baud
        self.timeout = timeout
        self._serial_factory = serial_factory
        self._serial: Any | None = None
        self.stream_command = stream_command
        self.stop_stream_command = stop_stream_command

    def __enter__(self) -> "SerialLuxNode":
        self.open()
        return self

    def __exit__(self, *_exc: object) -> None:
        self.close()

    def open(self) -> None:
        if self._serial is not None:
            return
        factory = self._serial_factory
        if factory is None:
            try:
                import serial  # type: ignore
            except ImportError as exc:
                raise LuxNodeError("Install serial support with: pip install 'luxspec[serial]'") from exc
            factory = serial.Serial
        self._serial = factory(self.port, self.baud, timeout=self.timeout)
        if hasattr(self._serial, "reset_input_buffer"):
            self._serial.reset_input_buffer()

    def close(self) -> None:
        if self._serial is not None and hasattr(self._serial, "close"):
            self._serial.close()
        self._serial = None

    def write_command(self, command: str) -> None:
        self.open()
        assert self._serial is not None
        self._serial.write((command.strip().upper() + "\n").encode("ascii"))

    def read_frame(self) -> LuxFrame:
        self.open()
        assert self._serial is not None
        while True:
            line = self._serial.readline()
            if isinstance(line, bytes):
                line = line.decode("utf-8", errors="replace")
            line = str(line).strip()
            if not line:
                raise LuxNodeError("Timed out waiting for a Luxnode JSON frame.")
            if line.startswith("{"):
                payload = json.loads(line)
                if "counts" in payload or "pixels" in payload:
                    return LuxFrame.from_mapping(payload)

    def capture(self, kind: str = "raw") -> LuxFrame:
        try:
            command = self.COMMANDS[kind.lower()]
        except KeyError as exc:
            raise ValueError(f"unknown capture kind: {kind}") from exc
        self.write_command(command)
        return self.read_frame()

    def start_streaming(self) -> None:
        if self.stream_command:
            self.write_command(self.stream_command)

    def stop_streaming(self) -> None:
        if self.stop_stream_command:
            self.write_command(self.stop_stream_command)

    def iter_frames(self) -> Iterator[LuxFrame]:
        self.start_streaming()
        try:
            while True:
                yield self.read_frame()
        finally:
            try:
                self.stop_streaming()
            except Exception:
                pass

luxspec.client.HttpLuxNode

Bases: BaseLuxNodeClient

HTTP client for Wi-Fi Luxnode firmware or a Lux Cloud edge bridge.

Source code in src/luxspec/client.py 
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
class HttpLuxNode(BaseLuxNodeClient):
    """HTTP client for Wi-Fi Luxnode firmware or a Lux Cloud edge bridge."""

    def __init__(self, base_url: str, *, timeout: float = 15.0) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def _json(self, path: str, *, method: str = "GET", body: dict[str, Any] | None = None) -> dict[str, Any]:
        data = None if body is None else json.dumps(body).encode("utf-8")
        headers = {"accept": "application/json"}
        if data is not None:
            headers["content-type"] = "application/json"
        req = request.Request(f"{self.base_url}{path}", data=data, headers=headers, method=method)
        try:
            with request.urlopen(req, timeout=self.timeout) as response:
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            raise LuxNodeError(f"Luxnode HTTP request failed with {exc.code}: {exc.reason}") from exc
        except URLError as exc:
            raise LuxNodeError(f"Could not reach Luxnode HTTP endpoint: {exc.reason}") from exc

    def capture(self, kind: str = "raw") -> LuxFrame:
        return LuxFrame.from_mapping(self._json("/api/v1/capture", method="POST", body={"kind": kind}))

    def telemetry(self) -> dict[str, Any]:
        return self._json("/api/v1/telemetry")

    def iter_frames(self) -> Iterator[LuxFrame]:
        while True:
            yield self.capture()

luxspec.processing.analyze_frame(frame, *, dark=None, reference=None, mode='reflectance', module=None, require_ready=True)

Process a raw frame into reflectance or absorbance and optional metrics.

Source code in src/luxspec/processing.py 
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
def analyze_frame(
    frame: LuxFrame,
    *,
    dark: LuxFrame | None = None,
    reference: LuxFrame | None = None,
    mode: str = "reflectance",
    module: str | None = None,
    require_ready: bool = True,
) -> FrameAnalysis:
    """Process a raw frame into reflectance or absorbance and optional metrics."""

    dark_counts = frame.dark_counts if frame.dark_counts is not None else (dark.counts if dark is not None else None)
    reference_counts = (
        frame.reference_counts if frame.reference_counts is not None else (reference.counts if reference is not None else None)
    )
    if dark_counts is None:
        raise ValueError("analyze_frame requires dark counts or a dark LuxFrame.")
    if reference_counts is None:
        raise ValueError("analyze_frame requires reference counts or a reference LuxFrame.")

    if mode == "reflectance":
        processed = process_reflectance(frame.wavelengths_nm, frame.counts, reference_counts, dark_counts, require_ready=require_ready)
    elif mode == "absorbance":
        processed = process_absorbance(frame.wavelengths_nm, frame.counts, reference_counts, dark_counts, require_ready=require_ready)
    else:
        raise ValueError("mode must be 'reflectance' or 'absorbance'.")

    return FrameAnalysis(processed=processed, metrics=module_metrics(processed.calibrated, module) if module else None)

luxspec.processing.module_metrics(spectrum, module)

Dispatch a calibrated spectrum to a Luxnode module metrics function.

Source code in src/luxspec/processing.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def module_metrics(spectrum, module: str | None):
    """Dispatch a calibrated spectrum to a Luxnode module metrics function."""

    if module is None:
        return None
    key = module.lower().replace("-", "_")
    if key in {"plant", "plants", "leaf"}:
        return plant_metrics(spectrum)
    if key in {"mineral", "minerals", "geology"}:
        return identify_mineral(spectrum)
    if key in {"beer", "brewing"}:
        return beer_metrics(spectrum)
    if key in {"kombucha", "fermentation"}:
        return kombucha_metrics(spectrum)
    if key in {"astronomy", "astro", "stellar", "stars", "telescope"}:
        return classify_stellar_spectrum(spectrum)
    raise ValueError(f"unknown Lux module: {module}")

luxspec.versioning.VersionInfo dataclass

Compatibility envelope reported by Luxnode firmware and Lux Cloud.

Source code in src/luxspec/versioning.py 
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
@dataclass(frozen=True)
class VersionInfo:
    """Compatibility envelope reported by Luxnode firmware and Lux Cloud."""

    firmware: SemVer
    protocol: SemVer
    hardware: str
    sdk_min: SemVer = SemVer(0, 2, 0)

    @classmethod
    def from_mapping(cls, payload: Mapping[str, Any]) -> "VersionInfo":
        return cls(
            firmware=SemVer.parse(str(payload["firmware"])),
            protocol=SemVer.parse(str(payload.get("protocol", "0.1.0"))),
            hardware=str(payload.get("hardware", "luxnode-v0")),
            sdk_min=SemVer.parse(str(payload.get("sdk_min", "0.2.0"))),
        )

    def supports_sdk(self, sdk_version: str) -> bool:
        sdk = SemVer.parse(sdk_version)
        return sdk.major == self.sdk_min.major and sdk >= self.sdk_min

    def supports_protocol(self, protocol_version: str) -> bool:
        protocol = SemVer.parse(protocol_version)
        return protocol.major == self.protocol.major and protocol <= self.protocol

luxspec.versioning.UpgradeManifest dataclass

Signed-release metadata for the Luxnode firmware updater.

Source code in src/luxspec/versioning.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass(frozen=True)
class UpgradeManifest:
    """Signed-release metadata for the Luxnode firmware updater."""

    firmware: SemVer
    protocol: SemVer
    hardware: str
    artifact_url: str
    artifact_sha256: str
    release_notes_url: str | None = None
    min_battery_mv: int = 3700

    @classmethod
    def from_mapping(cls, payload: Mapping[str, Any]) -> "UpgradeManifest":
        return cls(
            firmware=SemVer.parse(str(payload["firmware"])),
            protocol=SemVer.parse(str(payload.get("protocol", "0.1.0"))),
            hardware=str(payload.get("hardware", "luxnode-v0")),
            artifact_url=str(payload["artifact_url"]),
            artifact_sha256=str(payload["artifact_sha256"]).lower(),
            release_notes_url=payload.get("release_notes_url"),
            min_battery_mv=int(payload.get("min_battery_mv", 3700)),
        )

    def verify_file(self, path: str | Path) -> bool:
        digest = hashlib.sha256(Path(path).read_bytes()).hexdigest()
        return digest == self.artifact_sha256

lux.Spectrum dataclass

A one-dimensional spectrum with wavelengths in nanometers.

Source code in src/lux/spectrum.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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
67
68
69
70
71
72
73
74
75
@dataclass(frozen=True)
class Spectrum:
    """A one-dimensional spectrum with wavelengths in nanometers."""

    wavelengths_nm: np.ndarray
    values: np.ndarray
    kind: str = "intensity"
    metadata: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        wavelengths = np.asarray(self.wavelengths_nm, dtype=float)
        values = np.asarray(self.values, dtype=float)
        if wavelengths.ndim != 1 or values.ndim != 1:
            raise ValueError("wavelengths and values must be one-dimensional.")
        if wavelengths.size != values.size:
            raise ValueError("wavelength and value arrays must have the same length.")
        if wavelengths.size < 3:
            raise ValueError("a spectrum needs at least three channels.")
        object.__setattr__(self, "wavelengths_nm", wavelengths)
        object.__setattr__(self, "values", values)

    def interpolate_to(self, target_wavelengths_nm: np.ndarray) -> "Spectrum":
        target = np.asarray(target_wavelengths_nm, dtype=float)
        values = np.interp(target, self.wavelengths_nm, self.values)
        return Spectrum(target, values, kind=self.kind, metadata=dict(self.metadata))

    def window(self, min_nm: float, max_nm: float) -> "Spectrum":
        mask = (self.wavelengths_nm >= min_nm) & (self.wavelengths_nm <= max_nm)
        if mask.sum() < 3:
            raise ValueError("window leaves fewer than three channels.")
        return Spectrum(
            self.wavelengths_nm[mask],
            self.values[mask],
            kind=self.kind,
            metadata=dict(self.metadata),
        )

    def normalized(self, method: str = "l2") -> "Spectrum":
        y = self.values.astype(float).copy()
        if method == "l2":
            scale = np.linalg.norm(y)
            y = y / scale if scale > 0 else y
        elif method == "area":
            area = float(np.trapezoid(np.abs(y), self.wavelengths_nm))
            y = y / area if area > 0 else y
        elif method == "minmax":
            span = float(np.nanmax(y) - np.nanmin(y))
            y = (y - np.nanmin(y)) / span if span > 0 else y * 0
        elif method == "zscore":
            std = float(np.nanstd(y))
            y = (y - np.nanmean(y)) / std if std > 0 else y * 0
        else:
            raise ValueError(f"unknown normalization method: {method}")
        return Spectrum(self.wavelengths_nm, y, kind=self.kind, metadata=dict(self.metadata))

    def to_csv_rows(self) -> list[tuple[float, float]]:
        return list(zip(self.wavelengths_nm.tolist(), self.values.tolist()))

    @classmethod
    def from_pairs(
        cls,
        pairs: list[tuple[float, float]],
        kind: str = "intensity",
        metadata: Mapping[str, Any] | None = None,
    ) -> "Spectrum":
        arr = np.asarray(pairs, dtype=float)
        return cls(arr[:, 0], arr[:, 1], kind=kind, metadata=metadata or {})

lux.peaks.detect_peaks(wavelengths_nm, values, *, smooth_window=9, smooth_degree=3, min_prominence=None, min_distance_nm=10.0, baseline_correct=True, max_peaks=None)

Detect resolved peaks with local prominence and FWHM.

The default parameters are tuned for C12880MA/C16767MA spectra whose optical resolution is roughly 5-15 nm. Use smaller min_distance_nm only for the UV head or high-resolution external spectra.

Source code in src/lux/peaks.py 
 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
 67
 68
 69
 70
 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
110
def detect_peaks(
    wavelengths_nm: np.ndarray,
    values: np.ndarray,
    *,
    smooth_window: int = 9,
    smooth_degree: int = 3,
    min_prominence: float | None = None,
    min_distance_nm: float = 10.0,
    baseline_correct: bool = True,
    max_peaks: int | None = None,
) -> list[Peak]:
    """Detect resolved peaks with local prominence and FWHM.

    The default parameters are tuned for C12880MA/C16767MA spectra whose optical
    resolution is roughly 5-15 nm. Use smaller `min_distance_nm` only for the UV
    head or high-resolution external spectra.
    """

    wavelengths = np.asarray(wavelengths_nm, dtype=float)
    y = np.asarray(values, dtype=float)
    if wavelengths.size != y.size:
        raise ValueError("wavelength and value arrays must have the same length.")
    if wavelengths.size < smooth_window:
        smooth_window = wavelengths.size if wavelengths.size % 2 else wavelengths.size - 1
    smoothed = savitzky_golay(y, window=smooth_window, degree=min(smooth_degree, smooth_window - 1))
    if baseline_correct:
        baseline = asymmetric_least_squares_baseline(smoothed)
        signal = smoothed - baseline
    else:
        baseline = np.zeros_like(smoothed)
        signal = smoothed

    scaled = robust_scale(signal)
    robust_threshold = max(2.5, 0.05 * float(np.nanmax(scaled) - np.nanmin(scaled)))
    absolute_threshold = float(min_prominence) if min_prominence is not None else None

    candidate_indices: list[int] = []
    for i in range(1, signal.size - 1):
        if signal[i] >= signal[i - 1] and signal[i] > signal[i + 1]:
            left_min = float(np.min(signal[max(0, i - 20) : i + 1]))
            right_min = float(np.min(signal[i : min(signal.size, i + 21)]))
            prominence = signal[i] - max(left_min, right_min)
            robust_prominence = scaled[i] - max(np.min(scaled[max(0, i - 20) : i + 1]), np.min(scaled[i : min(signal.size, i + 21)]))
            passes_threshold = (
                prominence >= absolute_threshold
                if absolute_threshold is not None
                else robust_prominence >= robust_threshold
            )
            if prominence > 0 and passes_threshold:
                candidate_indices.append(i)

    candidate_indices.sort(key=lambda idx: signal[idx], reverse=True)
    selected: list[int] = []
    for idx in candidate_indices:
        if all(abs(wavelengths[idx] - wavelengths[other]) >= min_distance_nm for other in selected):
            selected.append(idx)
        if max_peaks is not None and len(selected) >= max_peaks:
            break
    selected.sort()

    peaks: list[Peak] = []
    for idx in selected:
        left_slice = signal[max(0, idx - 20) : idx + 1]
        right_slice = signal[idx : min(signal.size, idx + 21)]
        left_base_idx = max(0, idx - 20) + int(np.argmin(left_slice))
        right_base_idx = idx + int(np.argmin(right_slice))
        local_base = max(signal[left_base_idx], signal[right_base_idx])
        peaks.append(
            Peak(
                wavelength_nm=float(wavelengths[idx]),
                value=float(signal[idx]),
                prominence=float(signal[idx] - local_base),
                left_base_nm=float(wavelengths[left_base_idx]),
                right_base_nm=float(wavelengths[right_base_idx]),
                fwhm_nm=_half_width(wavelengths, signal, idx, local_base),
                index=int(idx),
            )
        )
    return peaks

lux.matching.match_spectrum(measured, library, *, top_k=5, smooth_window=9, metric='composite')

Rank spectra using complementary whole-shape metrics.

metric can be composite (the default), sam, correlation, derivative_correlation, or continuum_correlation.

Source code in src/lux/matching.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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def match_spectrum(
    measured: Spectrum,
    library: Iterable[LibraryEntry],
    *,
    top_k: int = 5,
    smooth_window: int = 9,
    metric: str = "composite",
) -> list[MatchResult]:
    """Rank spectra using complementary whole-shape metrics.

    ``metric`` can be ``composite`` (the default), ``sam``, ``correlation``,
    ``derivative_correlation``, or ``continuum_correlation``.
    """

    y = measured.values.astype(float)
    if smooth_window and y.size >= smooth_window:
        y = savitzky_golay(y, smooth_window, min(3, smooth_window - 1))
    y_deriv = first_derivative(measured.wavelengths_nm, y)
    y_cont = continuum_removed(y)

    results: list[MatchResult] = []
    for entry in library:
        ref = entry.spectrum.interpolate_to(measured.wavelengths_nm).values
        if smooth_window and ref.size >= smooth_window:
            ref = savitzky_golay(ref, smooth_window, min(3, smooth_window - 1))
        angle = spectral_angle(y, ref)
        corr = correlation(y, ref)
        dcorr = correlation(y_deriv, first_derivative(measured.wavelengths_nm, ref))
        ccorr = correlation(y_cont, continuum_removed(ref))
        angle_score = 1.0 - min(angle / np.pi, 1.0)
        if metric == "composite":
            score = 0.40 * angle_score + 0.30 * max(corr, -1) + 0.20 * max(dcorr, -1) + 0.10 * max(ccorr, -1)
        elif metric == "sam":
            score = angle_score
        elif metric == "correlation":
            score = 0.5 * (corr + 1.0)
        elif metric == "derivative_correlation":
            score = 0.5 * (dcorr + 1.0)
        elif metric in {"continuum", "continuum_removed", "continuum_correlation"}:
            score = 0.5 * (ccorr + 1.0)
        else:
            raise ValueError(f"unknown matching metric: {metric}")
        results.append(
            MatchResult(
                name=entry.name,
                group=entry.group,
                score=float(score),
                spectral_angle=float(angle),
                correlation=float(corr),
                derivative_correlation=float(dcorr),
                continuum_correlation=float(ccorr),
                metadata=dict(entry.metadata),
            )
        )
    results.sort(key=lambda item: item.score, reverse=True)
    return results[:top_k]

lux.modules.astronomy.astronomy_metrics(spectrum, *, color_calibration=None)

Compute V0 telescope-spectroscopy metrics from a calibrated spectrum.

Source code in src/lux/modules/astronomy.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
def astronomy_metrics(
    spectrum: Spectrum,
    *,
    color_calibration: ColorTemperatureCalibration | None = None,
) -> AstronomyMetrics:
    """Compute V0 telescope-spectroscopy metrics from a calibrated spectrum."""

    wavelengths = spectrum.wavelengths_nm
    values = np.asarray(spectrum.values, dtype=float)
    if np.nanmax(np.abs(values)) > 0:
        values = values / max(float(np.nanmax(np.abs(values))), 1e-12)

    h_alpha = _line_depth(wavelengths, values, BALMER_LINES_NM["H-alpha"])
    h_beta = _line_depth(wavelengths, values, BALMER_LINES_NM["H-beta"])
    h_gamma = _line_depth(wavelengths, values, BALMER_LINES_NM["H-gamma"])
    balmer = float(np.mean([h_alpha, h_beta, h_gamma]))
    o2_a = _line_depth(wavelengths, values, TELLURIC_LINES_NM["O2 A band"], line_half_width_nm=7.0, continuum_offset_nm=22.0)
    ratio = _blue_red_ratio(wavelengths, values)
    temperature = color_temperature_from_blue_red_ratio(ratio, calibration=color_calibration)

    flags: list[str] = []
    if wavelengths.min() > 430.0 or wavelengths.max() < 680.0:
        flags.append("missing-key-balmer-range")
    if float(np.nanmax(spectrum.values)) <= 1e-6:
        flags.append("low-signal")
    if float(np.nanpercentile(spectrum.values, 99)) >= 0.98 and spectrum.kind in {"reflectance", "relative_intensity"}:
        flags.append("possible-saturation")
    if o2_a > 0.12:
        flags.append("strong-telluric-oxygen-band")

    return AstronomyMetrics(
        blue_red_ratio=float(ratio),
        balmer_absorption_index=float(np.clip(balmer, -1.0, 1.0)),
        h_alpha_depth=float(h_alpha),
        h_beta_depth=float(h_beta),
        oxygen_a_depth=float(o2_a),
        continuum_slope_per_100nm=float(_continuum_slope(wavelengths, values)),
        approximate_color_temperature_k=float(temperature),
        probable_spectral_class=spectral_class_from_temperature(temperature),
        quality_flags=tuple(flags),
    )

lux.modules.astronomy.classify_stellar_spectrum(spectrum, *, library=None, color_calibration=None, top_k=5)

Classify a bright-object telescope spectrum against stellar templates.

Source code in src/lux/modules/astronomy.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
def classify_stellar_spectrum(
    spectrum: Spectrum,
    *,
    library: list[LibraryEntry] | None = None,
    color_calibration: ColorTemperatureCalibration | None = None,
    top_k: int = 5,
) -> StellarClassification:
    """Classify a bright-object telescope spectrum against stellar templates."""

    metrics = astronomy_metrics(spectrum, color_calibration=color_calibration)
    entries = library or stellar_template_library(spectrum.wavelengths_nm)
    matches = match_spectrum(spectrum, entries, top_k=top_k, smooth_window=7, metric="composite")
    return StellarClassification(metrics=metrics, matches=matches)

lux.modules.astronomy.fit_color_temperature_calibration(ratios, temperatures_k)

Fit a user calibration from known-star blue/red ratios and temperatures.

Source code in src/lux/modules/astronomy.py 
144
145
146
147
148
149
150
151
152
153
def fit_color_temperature_calibration(ratios: Iterable[float], temperatures_k: Iterable[float]) -> ColorTemperatureCalibration:
    """Fit a user calibration from known-star blue/red ratios and temperatures."""

    x = np.asarray(list(ratios), dtype=float)
    y = np.asarray(list(temperatures_k), dtype=float)
    mask = np.isfinite(x) & np.isfinite(y) & (x > 0) & (y > 0)
    if int(mask.sum()) < 2:
        raise ValueError("Need at least two positive known-star ratio/temperature pairs.")
    slope, intercept = np.polyfit(np.log(x[mask]), np.log(y[mask]), deg=1)
    return ColorTemperatureCalibration(intercept=float(intercept), slope=float(slope))

lux.modules.astronomy.stellar_template_library(wavelengths_nm)

Return a small broadened bright-star template library for V0 matching.

Source code in src/lux/modules/astronomy.py 
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
def stellar_template_library(wavelengths_nm: Iterable[float]) -> list[LibraryEntry]:
    """Return a small broadened bright-star template library for V0 matching."""

    w = np.asarray(list(wavelengths_nm), dtype=float)
    templates = [
        ("A0V Vega-like star", 9550.0, 0.30, 0.015, 0.00, "A0V"),
        ("F5V Procyon-like star", 6500.0, 0.17, 0.045, 0.00, "F5V"),
        ("G2V Sun-like star", 5770.0, 0.08, 0.075, 0.00, "G2V"),
        ("K5V orange star", 4400.0, 0.035, 0.115, 0.03, "K5V"),
        ("M3V red dwarf", 3400.0, 0.015, 0.10, 0.20, "M3V"),
    ]
    entries: list[LibraryEntry] = []
    for name, temp, balmer, metal, molecular, spectral_type in templates:
        values = _stellar_template_values(
            w,
            temperature_k=temp,
            balmer_strength=balmer,
            metal_strength=metal,
            molecular_strength=molecular,
        )
        entries.append(
            LibraryEntry(
                name=name,
                group="stellar_template",
                spectrum=Spectrum(w, values, kind="relative_intensity"),
                metadata={
                    "synthetic": "true",
                    "spectral_type": spectral_type,
                    "temperature_k": str(int(round(temp))),
                },
            )
        )
    return entries

lux.quality.assess_measurement_triplet(wavelengths_nm, sample_counts, reference_counts, dark_counts, *, mode, adc_max=65535.0, min_reference_signal=150.0)

Validate sample/reference/dark arrays before absorbance or reflectance.

Source code in src/lux/quality.py 
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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def assess_measurement_triplet(
    wavelengths_nm: Iterable[float],
    sample_counts: Iterable[float],
    reference_counts: Iterable[float],
    dark_counts: Iterable[float],
    *,
    mode: str,
    adc_max: float = 65535.0,
    min_reference_signal: float = 150.0,
) -> QualityReport:
    """Validate sample/reference/dark arrays before absorbance or reflectance."""

    w = _as_array(wavelengths_nm, "wavelengths_nm")
    sample = _as_array(sample_counts, "sample_counts")
    reference = _as_array(reference_counts, "reference_counts")
    dark = _as_array(dark_counts, "dark_counts")
    checks: list[QualityCheck] = []
    checks.extend(assess_wavelength_axis(w).checks)
    checks.extend(assess_counts(sample, name="sample_counts", adc_max=adc_max).checks)
    checks.extend(assess_counts(reference, name="reference_counts", adc_max=adc_max).checks)
    checks.extend(assess_counts(dark, name="dark_counts", adc_max=adc_max, min_peak_count=1.0, min_dynamic_range=0.0).checks)

    same_shape = sample.shape == reference.shape == dark.shape == w.shape
    checks.append(
        _check(
            "array_shapes_match",
            same_shape,
            "error",
            "Sample, reference, dark, and wavelength arrays must have the same length.",
            value=str((sample.shape, reference.shape, dark.shape, w.shape)),
        )
    )
    if not same_shape:
        return QualityReport(tuple(checks))

    reference_signal = reference - dark
    sample_signal = sample - dark
    valid_reference_fraction = float(np.mean(reference_signal > min_reference_signal))
    valid_sample_fraction = float(np.mean(sample_signal > 1.0))
    checks.append(
        _check(
            "reference_above_dark",
            valid_reference_fraction >= 0.90,
            "error",
            "Reference/blank is too close to dark current over too many channels.",
            value=valid_reference_fraction,
            threshold=">= 0.90",
        )
    )
    checks.append(
        _check(
            "sample_above_dark",
            valid_sample_fraction >= 0.70,
            "warning",
            "Sample is close to dark current; increase integration time or illumination.",
            value=valid_sample_fraction,
            threshold=">= 0.70",
        )
    )

    corrected_ratio = np.clip(sample_signal, 1e-12, None) / np.clip(reference_signal, 1e-12, None)
    if mode == "reflectance":
        median_ratio = float(np.nanmedian(corrected_ratio))
        checks.append(
            _check(
                "reflectance_reference_geometry",
                median_ratio <= 1.25,
                "warning",
                "Median reflectance exceeds reference by more than expected; check geometry/reference tile.",
                value=median_ratio,
                threshold="<= 1.25",
            )
        )
    elif mode == "absorbance":
        high_transmission = float(np.nanpercentile(corrected_ratio, 95))
        checks.append(
            _check(
                "absorbance_blank_geometry",
                high_transmission <= 1.10,
                "warning",
                "Sample transmission exceeds blank in high percentile channels; verify blank/sample pairing.",
                value=high_transmission,
                threshold="<= 1.10",
            )
        )
    else:
        checks.append(_check("mode_supported", False, "error", "Mode must be 'reflectance' or 'absorbance'.", value=mode))

    return QualityReport(tuple(checks))

luxspec.prototype.capture_prototype_measurement(client, *, mode='reflectance', module=None, sample_kind='raw', reference_kind=None, dark_kind='dark', require_ready=True)

Capture dark/reference/sample frames and run the full V0 analysis path.

This is the one-call API for bench use: it captures the required frames, validates signal quality, processes absorbance or reflectance, and dispatches module metrics.

Source code in src/luxspec/prototype.py 
 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
def capture_prototype_measurement(
    client: BaseLuxNodeClient,
    *,
    mode: str = "reflectance",
    module: str | None = None,
    sample_kind: str = "raw",
    reference_kind: str | None = None,
    dark_kind: str = "dark",
    require_ready: bool = True,
) -> PrototypeMeasurement:
    """Capture dark/reference/sample frames and run the full V0 analysis path.

    This is the one-call API for bench use: it captures the required frames,
    validates signal quality, processes absorbance or reflectance, and dispatches
    module metrics.
    """

    if reference_kind is None:
        reference_kind = "white" if mode == "reflectance" else "blank"
    dark = client.capture(dark_kind)
    reference = client.capture(reference_kind)
    sample = client.capture(sample_kind)
    quality = assess_measurement_triplet(
        sample.wavelengths_nm,
        sample.counts,
        reference.counts,
        dark.counts,
        mode=mode,
    )
    if require_ready:
        quality.raise_for_errors()
    analysis = analyze_frame(sample, dark=dark, reference=reference, mode=mode, module=module, require_ready=require_ready)
    return PrototypeMeasurement(mode=mode, module=module, dark=dark, reference=reference, sample=sample, analysis=analysis, quality=quality)

lux.experiments.run_experiment(scenario_id)

Run one deterministic synthetic experiment.

Parameters:

Name Type Description Default
scenario_id str

Identifier from :func:scenario_ids.

 required

Returns:

Type Description
dict[str, Any]

JSON-serializable dashboard payload with spectra, metrics, peaks,
dict[str, Any]

matches, and real-data replacement notes.
Source code in src/lux/experiments.py 
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
def run_experiment(scenario_id: str) -> dict[str, Any]:
    """Run one deterministic synthetic experiment.

    Args:
        scenario_id: Identifier from :func:`scenario_ids`.

    Returns:
        JSON-serializable dashboard payload with spectra, metrics, peaks,
        matches, and real-data replacement notes.
    """

    _init_scenarios()
    try:
        return SCENARIOS[scenario_id].run()
    except KeyError as exc:
        known = ", ".join(scenario_ids())
        raise ValueError(f"unknown synthetic experiment {scenario_id!r}. Known ids: {known}") from exc

lux.experiments.run_all_experiments()

Run every registered synthetic experiment.

Source code in src/lux/experiments.py 
1146
1147
1148
1149
1150
def run_all_experiments() -> list[dict[str, Any]]:
    """Run every registered synthetic experiment."""

    _init_scenarios()
    return [SCENARIOS[scenario_id].run() for scenario_id in scenario_ids()]

lux.experiments.write_experiments_json(path)

Write the docs dashboard payload to path.

Source code in src/lux/experiments.py 
1165
1166
1167
1168
1169
1170
def write_experiments_json(path: str | Path) -> None:
    """Write the docs dashboard payload to ``path``."""

    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(experiments_payload(), indent=2) + "\n")