fujilib.protocol¶

def as_decimal(raw: int, decimals: int) -> Decimal:
    """Return the exact decimal value of ``raw`` at ``decimals`` places.

    Raises:
        FujiDecodeError: ``decimals`` is outside 0–3.
    """
    _check_decimals(decimals)
    return Decimal(raw).scaleb(-decimals)

def decode_bcd(word: int) -> int:
    """Decode four BCD digits (``0x2359`` → 2359).

    Raises:
        FujiDecodeError: a nibble is above 9, or ``word`` is not a 16-bit value.
    """
    _check_words((word,), 1, "BCD")
    value = 0
    for shift in (12, 8, 4, 0):
        digit = (word >> shift) & 0xF
        if digit > 9:  # noqa: PLR2004
            msg = f"0x{word:04X} is not a BCD value"
            raise FujiDecodeError(msg, context=_ctx(word=word))
        value = value * 10 + digit
    return value

def decode_bool(word: int) -> bool:
    """Decode a whole-register flag: 0 is off, anything else on.

    Any non-zero value counts as set, so an unexpected value in a status flag
    errs towards "held" or "calibrating", which marks data invalid rather than
    valid.

    Raises:
        FujiDecodeError: ``word`` is not a 16-bit value.
    """
    _check_words((word,), 1, "bool")
    return word != 0

def decode_chars(words: Sequence[int], *, strip: bool = True) -> str:
    """Decode one ASCII character per register, from the low byte (design §2.5).

    A zero low byte is a blank and becomes a space, so positions are kept (the
    type code is decoded by digit position). With ``strip`` the result has
    trailing blanks removed.

    Raises:
        FujiDecodeError: a high byte is non-zero or a character is not printable
            ASCII. Either means the block was not a character field, which a
            probe treats as invalid data rather than a string.
    """
    _check_words(words, None, "characters")
    chars: list[str] = []
    for word in words:
        high, low = word >> 8, word & 0xFF
        if high != 0 or not (low == 0 or low in _ASCII_PRINTABLE):
            msg = f"register 0x{word:04X} is not one ASCII character in the low byte"
            raise FujiDecodeError(msg, context=_ctx(words=tuple(words)))
        chars.append(" " if low == 0 else chr(low))
    text = "".join(chars)
    return text.rstrip() if strip else text

def decode_enum[E: IntEnum](enum: type[E], raw: int, *, strict: bool = False) -> E | int:
    """Decode ``raw`` as a member of ``enum``.

    A value the manual does not define is returned as a plain ``int`` so it is
    kept, not lost; with ``strict`` it raises instead.

    Raises:
        FujiDecodeError: ``strict`` and ``raw`` is not a member of ``enum``.
    """
    try:
        return enum(raw)
    except ValueError:
        if strict:
            msg = f"{raw!r} is not a documented {enum.__name__} value"
            raise FujiDecodeError(msg, context=_ctx(value=raw)) from None
        return raw

def decode_int(word: int, *, signed: bool) -> int:
    """Decode one register as a 16-bit integer (two's complement when ``signed``).

    Raises:
        FujiDecodeError: ``word`` is not a 16-bit value.
    """
    _check_words((word,), 1, "int16" if signed else "uint16")
    return decode_int16((word,), signed=signed, byte_order=ByteOrder.BIG)

def decode_raw(words: Sequence[int], data_type: DataType) -> int | bool | str:
    """Decode ``words`` by ``data_type`` without scaling or enum lookup.

    :attr:`DataType.ENUM` returns the raw code. Scaling and enum meaning need
    the register's spec and live above the codec.

    Raises:
        FujiDecodeError: wrong width or an undecodable value.
    """
    width = data_type.fixed_width
    _check_words(words, width, data_type.value)
    match data_type:
        case DataType.UINT16 | DataType.ENUM:
            return decode_int(words[0], signed=False)
        case DataType.INT16:
            return decode_int(words[0], signed=True)
        case DataType.UINT32_LH:
            return decode_uint32_lh(words)
        case DataType.BCD:
            return decode_bcd(words[0])
        case DataType.BOOL:
            return decode_bool(words[0])
        case DataType.CHAR:
            return decode_chars(words)
        case _:  # pragma: no cover — every DataType has a case
            assert_never(data_type)

def decode_uint32_lh(words: Sequence[int]) -> int:
    """Decode a long word: two registers, low-order word first (design §2.5).

    Raises:
        FujiDecodeError: not exactly two 16-bit words.
    """
    _check_words(words, 2, "uint32 (low word first)")
    return decode_int32(
        tuple(words), signed=False, word_order=WordOrder.LOW_HIGH, byte_order=ByteOrder.BIG
    )

def encode_bcd(value: int) -> int:
    """Encode 0–9999 as four BCD digits (2359 → ``0x2359``).

    Raises:
        FujiValidationError: ``value`` is outside 0–9999.
    """
    if not 0 <= value <= _BCD_MAX:
        msg = f"{value!r} is outside the BCD range 0-{_BCD_MAX}"
        raise FujiValidationError(msg, context=_ctx(value=value))
    word = 0
    for shift, digit in zip((12, 8, 4, 0), f"{value:04d}", strict=True):
        word |= int(digit) << shift
    return word

def encode_chars(text: str, width: int) -> tuple[int, ...]:
    """Encode ``text`` as one character per register, padded with blanks to ``width``.

    Raises:
        FujiValidationError: ``text`` is longer than ``width`` or not printable ASCII.
    """
    if len(text) > width or any(ord(c) not in _ASCII_PRINTABLE for c in text):
        msg = f"{text!r} is not up to {width} printable ASCII characters"
        raise FujiValidationError(msg, context=_ctx(text=text))
    return tuple(ord(c) for c in text) + (0,) * (width - len(text))

def encode_int(value: int, *, signed: bool) -> int:
    """Encode ``value`` as one register word.

    Raises:
        FujiValidationError: ``value`` does not fit in 16 bits.
    """
    try:
        (word,) = encode_int16(value, signed=signed, byte_order=ByteOrder.BIG)
    except (ValueError, TypeError, OverflowError) as exc:
        kind = "int16" if signed else "uint16"
        msg = f"{value!r} does not fit in an {kind} register"
        raise FujiValidationError(msg, context=_ctx(value=value)) from exc
    return word

def encode_uint32_lh(value: int) -> tuple[int, int]:
    """Encode ``value`` as a long word, low-order word first.

    Raises:
        FujiValidationError: ``value`` does not fit in 32 unsigned bits.
    """
    try:
        return encode_int32(
            value, signed=False, word_order=WordOrder.LOW_HIGH, byte_order=ByteOrder.BIG
        )
    except (ValueError, TypeError, OverflowError) as exc:
        msg = f"{value!r} does not fit in an unsigned long word"
        raise FujiValidationError(msg, context=_ctx(value=value)) from exc

def scale(raw: int, decimals: int) -> float:
    """Apply a decimal-point position: ``scale(2029, 2)`` → ``20.29``.

    Always returns a ``float``, so a concentration column never flips between
    ``int`` and ``float`` when the decimal point is 0.

    Raises:
        FujiDecodeError: ``decimals`` is outside 0–3.
    """
    _check_decimals(decimals)
    return raw / _POWERS_OF_TEN[decimals]

def unscale(value: float | int | Decimal | str, decimals: int) -> int:
    """Invert :func:`scale`: ``unscale(20.29, 2)`` → ``2029``.

    Raises:
        FujiValidationError: ``value`` is not finite, or has more precision
            than ``decimals`` places can hold.
        FujiDecodeError: ``decimals`` is outside 0–3.
    """
    _check_decimals(decimals)
    try:
        exact = Decimal(repr(value)) if isinstance(value, float) else Decimal(value)
    except (InvalidOperation, ValueError) as exc:
        msg = f"{value!r} is not a number"
        raise FujiValidationError(msg, context=_ctx(value=value)) from exc
    if not exact.is_finite():
        msg = f"{value!r} is not finite"
        raise FujiValidationError(msg, context=_ctx(value=value))
    shifted = exact.scaleb(decimals)
    rounded = shifted.to_integral_value(rounding=ROUND_HALF_EVEN)
    if abs(shifted - rounded) > _UNSCALE_TOLERANCE:
        msg = f"{value!r} has more than {decimals} decimal place(s)"
        raise FujiValidationError(msg, context=_ctx(value=value, decimals=decimals))
    return int(rounded)

def calibration_log_plan(channel: int) -> tuple[BlockRead, ...]:
    """The block reads of one channel's calibration log (design §4.3)."""
    return plan_log_reads(CALIBRATION_LOG, channel)

def plan_log_reads(
    log: LogSpec,
    channel: int = 1,
    *,
    policy: ReadPolicy = DEFAULT_READ_POLICY,
) -> tuple[BlockRead, ...]:
    """Plan reads of one channel's region of ``log``, in whole records.

    Each block holds as many whole records as fit, so a block ends exactly at
    a record boundary (the calibration log reads as 5 × 63 + 45 words).

    Raises:
        FujiValidationError: ``channel`` is not a channel of ``log``.
        FujiConfigurationError: one record is wider than a request.
    """
    per_block = policy.max_words // log.record_words
    if per_block < 1:
        msg = f"a {log.name} record does not fit in one request"
        raise FujiConfigurationError(msg)
    fc = log.table.read_function
    blocks: list[BlockRead] = []
    for first in range(0, log.records, per_block):
        records = min(per_block, log.records - first)
        address = log.record_address(channel, first)
        blocks.append(BlockRead(function=fc, address=address, count=records * log.record_words))
    return tuple(blocks)

def plan_reads(
    specs: Iterable[RegisterSpec],
    *,
    regions: RegionMap = ZP_REGIONS,
    policy: ReadPolicy = DEFAULT_READ_POLICY,
) -> tuple[BlockRead, ...]:
    """Plan the fewest block reads covering ``specs``.

    Blocks come out holding table first, then input, each in address order.

    Raises:
        FujiConfigurationError: a register lies outside every region, or is
            wider than one request.
    """
    unique = {s.name: s for s in specs}.values()
    blocks: list[BlockRead] = []
    for table in RegisterTable:
        fc = table.read_function
        ordered = sorted((s for s in unique if s.table is table), key=lambda s: s.address)
        run: list[RegisterSpec] = []
        run_region = None
        for spec in ordered:
            region = regions.region_for(fc, spec.address, spec.count)
            if region is None or spec.count > policy.max_words:
                msg = f"{spec.name} cannot be read in one FC{fc:02X} request inside a region"
                raise FujiConfigurationError(
                    msg, context=ErrorContext(function_code=fc, register_address=spec.address)
                )
            if run:
                start, end = run[0].address, max(s.last_address for s in run)
                gap = spec.address - end - 1
                merged = max(end, spec.last_address) - start + 1
                if region is run_region and gap <= policy.max_gap and merged <= policy.max_words:
                    run.append(spec)
                    continue
                blocks.append(_block(fc, run))
            run, run_region = [spec], region
        if run:
            blocks.append(_block(fc, run))
    return tuple(blocks)