slmm/nl52/protocol.py

"""
NL-42 / NL-52 serial protocol — pure parsing, no I/O.

Reference: RION NL-42/NL-52 Serial Interface Manual (No. 55779).

Command grammar (shared with the NL-43 family):
    Setting:  "$" + name + "," + param + CRLF      e.g. "$Measure,Start"
    Request:  name + "?" + CRLF                     e.g. "DOD?"
    Reply:    result code "R+0000" + CRLF, then for requests a data line.
    Stop DRD stream: SUB (0x1A).

This module is intentionally I/O-free so it can be unit-tested without a
device and reused unchanged behind any transport (serial, or a serial->TCP
bridge).
"""

from dataclasses import dataclass, field
from typing import Dict, List, Optional

CR = "\r"
LF = "\n"
CRLF = "\r\n"
SUB = b"\x1a"  # stop DRD streaming

# Result codes (Serial Interface Manual, "Result code")
RESULT_CODES: Dict[str, str] = {
    "R+0000": "Normal end",
    "R+0001": "Command error (command not recognized)",
    "R+0002": "Parameter error (bad count/type of parameters)",
    "R+0003": "Designation error (setting sent to request-only cmd, or vice versa)",
    "R+0004": "Status error (command not valid in current state)",
}

# DOD? response field order (d1..d14). Main channel unless noted.
DOD_FIELDS: List[str] = [
    "lp", "leq", "le", "lmax", "lmin",
    "ly",                       # additional processing value (e.g. LCpeak)
    "ln1", "ln2", "ln3", "ln4", "ln5",
    "sub_lp",                   # sub channel Lp
    "overload", "underrange",   # 0/1 flags
]

# DRD? response field order (d0..d8). Requires NX-42EX.
DRD_FIELDS: List[str] = [
    "counter",                  # d0: 1..600
    "lp", "leq", "lmax", "lmin",
    "ly",
    "sub_lp",
    "overload", "underrange",
]

# Token the meter returns for a disabled/unavailable display channel.
_NULL_TOKENS = {"--.-", "-.-", "---.-", ""}


class ResultError(Exception):
    """Raised when the meter returns a non-OK result code (R+0001..0004)."""

    def __init__(self, code: str):
        self.code = code
        self.meaning = RESULT_CODES.get(code, "Unknown result code")
        super().__init__(f"{code}: {self.meaning}")


def is_result_code(line: str) -> bool:
    line = line.strip()
    return line.startswith("R+") and len(line) == 6 and line[2:].isdigit()


def parse_level(token: str) -> Optional[float]:
    """Parse a space-padded level token into a float, or None if disabled.

    The meter pads to 5 chars and returns '--.-' (leading space) for channels
    whose display is OFF.
    """
    t = token.strip()
    if t in _NULL_TOKENS or set(t) <= set("-. "):
        return None
    try:
        return float(t)
    except ValueError:
        return None


def _parse_flag(token: str) -> Optional[bool]:
    t = token.strip()
    if t == "1":
        return True
    if t == "0":
        return False
    return None


@dataclass
class DODSnapshot:
    """Parsed DOD? snapshot (displayed values). Levels are dB or None if the
    channel's display is OFF. Measurement state is NOT part of DOD — query
    Measure? separately."""

    lp: Optional[float] = None
    leq: Optional[float] = None
    le: Optional[float] = None
    lmax: Optional[float] = None
    lmin: Optional[float] = None
    ly: Optional[float] = None
    ln1: Optional[float] = None
    ln2: Optional[float] = None
    ln3: Optional[float] = None
    ln4: Optional[float] = None
    ln5: Optional[float] = None
    sub_lp: Optional[float] = None
    overload: Optional[bool] = None
    underrange: Optional[bool] = None
    raw: str = ""


@dataclass
class DRDSample:
    """Parsed DRD? sample (continuous output, ~every 100 ms)."""

    counter: Optional[int] = None
    lp: Optional[float] = None
    leq: Optional[float] = None
    lmax: Optional[float] = None
    lmin: Optional[float] = None
    ly: Optional[float] = None
    sub_lp: Optional[float] = None
    overload: Optional[bool] = None
    underrange: Optional[bool] = None
    raw: str = ""


def _split(resp: str) -> List[str]:
    return [p for p in resp.strip().split(",")]


def parse_dod(resp: str) -> DODSnapshot:
    """Parse a DOD? data line into a DODSnapshot.

    Tolerant of trailing/short field counts — only maps what is present.
    """
    parts = _split(resp)
    snap = DODSnapshot(raw=resp.strip())

    for idx, name in enumerate(DOD_FIELDS):
        if idx >= len(parts):
            break
        if name in ("overload", "underrange"):
            setattr(snap, name, _parse_flag(parts[idx]))
        else:
            setattr(snap, name, parse_level(parts[idx]))

    return snap


def parse_drd(resp: str) -> DRDSample:
    """Parse a single DRD? data line into a DRDSample."""
    parts = _split(resp)
    sample = DRDSample(raw=resp.strip())

    for idx, name in enumerate(DRD_FIELDS):
        if idx >= len(parts):
            break
        if name == "counter":
            t = parts[idx].strip()
            sample.counter = int(t) if t.isdigit() else None
        elif name in ("overload", "underrange"):
            setattr(sample, name, _parse_flag(parts[idx]))
        else:
            setattr(sample, name, parse_level(parts[idx]))

    return sample