fujilib.registry¶

def validate_map(
    specs: Sequence[RegisterSpec],
    logs: Sequence[LogSpec],
    regions: RegionMap,
) -> None:
    """Validate a register map; see the module docstring for the rules.

    Takes the map as arguments so tests can feed it deliberately broken tables.

    Raises:
        FujiConfigurationError: the first rule the map breaks.
    """
    names: set[str] = set()
    spans: list[tuple[RegisterTable, int, int, str]] = []
    for spec in specs:
        if spec.name in names:
            _fail(f"duplicate register name {spec.name!r}", spec.table, spec.address)
        names.add(spec.name)
        _check_spec(spec, regions)
        spans.append((spec.table, spec.address, spec.last_address, spec.name))
    for log in logs:
        if log.name in names:
            _fail(f"duplicate name {log.name!r}", log.table, log.base)
        names.add(log.name)
        size = log.last_address - log.base + 1
        if regions.region_for(log.table.read_function, log.base, size) is None:
            _fail(f"{log.name}: not inside one region", log.table, log.base)
        for fld in log.fields:
            if fld.offset + fld.count > log.record_words:
                _fail(f"{log.name}.{fld.name}: outside the record", log.table, log.base)
        spans.append((log.table, log.base, log.last_address, log.name))
    spans.sort()
    for before, after in pairwise(spans):
        if before[0] is after[0] and after[1] <= before[2]:
            _fail(f"{before[3]} and {after[3]} overlap", after[0], after[1])

def check_envelope(
    fc: int, address: int, count: int = 1, *, values: Sequence[int] | None = None
) -> None:
    """Refuse a write outside :data:`WRITE_ENVELOPE`.

    Raises:
        FujiValidationError: the write is outside the envelope.
    """
    if not envelope_allows(fc, address, count, values=values):
        shown = "" if values is None else f" of {', '.join(f'0x{v:04X}' for v in values)}"
        msg = (
            f"FC{fc:02X} write of {count} word(s){shown} at 0x{address:04X} is outside the "
            "write envelope"
        )
        extra: dict[str, object] = {"count": count}
        if values is not None:
            extra["values"] = tuple(values)
        raise FujiValidationError(
            msg,
            context=ErrorContext(function_code=fc, register_address=address, extra=extra),
        )

def envelope_allows(
    fc: int, address: int, count: int = 1, *, values: Sequence[int] | None = None
) -> bool:
    """Whether a ``count``-word write at ``address`` with ``fc`` lies inside one envelope range.

    Where a range limits the words written (key simulation), ``values`` must
    be given and each be one of them: an address alone is not enough there.
    """
    return any(r.allows(fc, address, count, values) for r in WRITE_ENVELOPE)

def channel_layout(
    ndir: Sequence[Gas],
    *,
    o2: bool,
    correction: O2Correction,
    source: LabelSource,
) -> ChannelLayout:
    """Apply the ZPA manual's layout rule (§5.3(3)); see the module docstring.

    Correction needs an O2 value, so without ``o2`` no corrected channels are
    laid out.
    """
    corrected = correction.has_corrected or correction.has_average
    shown = tuple(Gas.NOX if g is Gas.NO and corrected and o2 else g for g in ndir)
    out: list[ChannelSuggestion] = []

    def add(gas: Gas, role: ChannelRole, derived_from: ChannelId | None = None) -> ChannelId:
        channel = ChannelId.from_number(len(out) + 1)
        out.append(ChannelSuggestion(channel, gas, role, source, derived_from))
        return channel

    component_channel = {gas: add(gas, ChannelRole.INSTANTANEOUS) for gas in shown}
    if o2:
        add(Gas.O2, ChannelRole.INSTANTANEOUS)
        targets = [g for g, orig in zip(shown, ndir, strict=True) if orig in _CORRECTABLE]
        if correction.has_corrected:
            for gas in targets:
                add(gas, ChannelRole.O2_CORRECTED, component_channel[gas])
        if correction.has_average:
            for gas in targets:
                add(gas, ChannelRole.O2_CORRECTED_AVERAGE, component_channel[gas])
    return ChannelLayout(tuple(out))

def decode_type_code(
    raw: str,
    decoders: Mapping[str, TypeCodeDecoder] = TYPECODE_DECODERS,
) -> TypeCode:
    """Decode ``raw`` with the table for its model, or keep it raw.

    Never raises. A code whose first three characters name no known model
    comes back with only ``raw`` (and ``model`` when it starts with ``ZP``).
    """
    text = raw.rstrip()
    model = text[:3]
    decoder = decoders.get(model)
    if decoder is None:
        return TypeCode(raw=text, model=model if model.startswith("ZP") else None)
    return decoder.decode(text)

def suggest_labels(
    type_code: TypeCode | None,
    present: Iterable[ChannelId],
) -> Mapping[ChannelId, ChannelSuggestion]:
    """Suggest a label for each present channel (design §2.9, step 4).

    A channel the type-code layout covers gets its suggestion. A present
    channel just after the layout's NDIR components, when the layout has no
    O2, is suggested as O2 by the layout rule, labelled ``INFERRED``: this is
    the bench unit, whose code says "no O2" while channel 3 measures O2. Any
    other present channel gets no suggestion. A suggestion never selects a
    scientific channel by itself; only an asserted label does.
    """
    layout = type_code.layout if type_code is not None else None
    suggestions: dict[ChannelId, ChannelSuggestion] = {}
    if layout is None:
        return MappingProxyType(suggestions)
    after_ndir = ChannelId.from_number(len(layout.channels) + 1) if layout.channels else None
    for channel in present:
        suggestion = layout.get(channel)
        if suggestion is not None:
            suggestions[channel] = suggestion
        elif layout.o2_channel is None and channel is after_ndir and not _has_derived(layout):
            suggestions[channel] = ChannelSuggestion(
                channel, Gas.O2, ChannelRole.INSTANTANEOUS, LabelSource.INFERRED
            )
    return MappingProxyType(suggestions)

def coerce_channel(channel: ChannelId | str) -> ChannelId:
    """Resolve ``channel`` to a :class:`ChannelId`.

    Accepts a :class:`ChannelId` or a string such as ``"CH3"``, ``"ch3"`` or
    ``"3"``.

    Raises:
        FujiValidationError: ``channel`` names no channel.
    """
    if isinstance(channel, ChannelId):
        return channel
    text = channel.strip().upper()
    if text.isdigit():
        return ChannelId.from_number(int(text))
    try:
        return ChannelId(text)
    except ValueError:
        msg = f"unknown channel {channel!r}; expected CH1-CH12"
        raise FujiValidationError(msg, context=ErrorContext(channel=channel)) from None

def coerce_channel_map(
    channel_map: Mapping[ChannelId | str, Gas | str],
) -> Mapping[ChannelId, Gas]:
    """Resolve a caller's channel map, as ``open_device`` and ``identify()`` take it.

    Raises:
        FujiValidationError: a channel or gas is unknown, a channel is asserted
            as ``unknown``, or one channel is named twice with different gases.
    """
    out: dict[ChannelId, Gas] = {}
    for key, value in channel_map.items():
        channel, gas = coerce_channel(key), coerce_gas(value)
        if gas is Gas.UNKNOWN:
            msg = f"{channel.value} cannot be asserted as unknown; leave it out instead"
            raise FujiValidationError(msg, context=ErrorContext(channel=channel.value))
        if out.setdefault(channel, gas) is not gas:
            msg = f"{channel.value} is asserted as both {out[channel].value} and {gas.value}"
            raise FujiValidationError(msg, context=ErrorContext(channel=channel.value))
    return MappingProxyType(dict(sorted(out.items(), key=lambda item: item[0].number)))

def coerce_gas(gas: Gas | str) -> Gas:
    """Resolve ``gas`` to a :class:`Gas`, from a member or its formula in any case (``"O2"``).

    Raises:
        FujiValidationError: ``gas`` names no gas.
    """
    if isinstance(gas, Gas):
        return gas
    try:
        return Gas(gas.strip().lower())
    except ValueError:
        known = ", ".join(g.value for g in Gas if g is not Gas.UNKNOWN)
        msg = f"unknown gas {gas!r}; expected one of {known}"
        raise FujiValidationError(msg) from None

def coerce_unit(unit: Unit | str) -> Unit:
    """Map a :class:`Unit` or a unit string (case-insensitive) to a :class:`Unit`.

    Unrecognised text becomes :attr:`Unit.UNKNOWN`.
    """
    if isinstance(unit, Unit):
        return unit
    return _BY_TEXT.get(unit.strip().casefold(), Unit.UNKNOWN)

def unit_code(unit: Unit) -> int:
    """Return the register code for ``unit``.

    Raises:
        FujiValidationError: ``unit`` is :attr:`Unit.UNKNOWN`, which has no code.
    """
    try:
        return _CODE_OF[unit]
    except KeyError:
        msg = f"unit {unit.value!r} has no register code"
        raise FujiValidationError(msg) from None

def unit_from_code(code: int) -> Unit:
    """Map a unit-code register value (0 vol%, 1 ppm, 2 mg/m³, 3 g/m³) to a :class:`Unit`.

    Any other value becomes :attr:`Unit.UNKNOWN`.
    """
    return _BY_CODE.get(code, Unit.UNKNOWN)