fujilib.devices¶

async def apply_settings(
    self,
    document: SettingsDocument | Mapping[str, object],
    *,
    confirm: bool = False,
    any_analyzer: bool = False,
    max_tier: SafetyTier = SafetyTier.DANGEROUS,
    timeout: float | None = None,
) -> ApplyReport:
    """Write the settings of a document that differ from the analyzer's.

    The whole document is compared first (:meth:`diff_settings`); if any
    setting is refused, nothing is written. ``confirm`` must then be
    ``True``; the CLI also asks for its destructive flag when a write is
    DANGEROUS. The writes go one at a time in dependency order, each read
    back; the first that fails stops the rest, and nothing is rolled back.
    ``max_tier`` refuses a document whose writes go above it, judged on
    the comparison made here, not on an earlier one: the CLI passes
    ``PERSISTENT`` unless its destructive flag is given. ``timeout`` bounds
    the whole apply.

    Raises:
        FujiValidationError: the document is not a settings document, or is
            refused; nothing was written.
        FujiConfirmationRequiredError: there is something to write and
            ``confirm`` is not ``True``, or a write is above ``max_tier``;
            nothing was written.
        FujiError: the comparison's reads failed. A failed *write* does not
            raise: it is in the report.
    """
    deadline = Deadline.after(timeout, operation="apply_settings")
    diff = await self.diff_settings(
        document, any_analyzer=any_analyzer, timeout=_remaining(deadline)
    )
    if not diff.ok:
        raise diff.refusal()
    if not diff.writes:
        return ApplyReport(diff, ())
    if diff.tier > max_tier:
        msg = (
            f"applying the settings would make {diff.tier.name} writes, above "
            f"{max_tier.name}; nothing was written"
        )
        raise FujiConfirmationRequiredError(
            msg, context=ErrorContext(extra={"safety": diff.tier.name.lower()})
        )
    self._session.gate(
        "apply_settings", tier=diff.tier, confirm=confirm, subject="applying the settings"
    )
    registry = self._session.profile.registry
    completed: list[WriteResult] = []
    writes = diff.writes
    for index, change in enumerate(writes):
        spec = registry.resolve(change.name)
        try:
            result = await self._write(
                "apply_settings",
                spec,
                change.desired.value,
                unit=change.desired.unit,
                confirm=confirm,
                timeout=_remaining(deadline),
            )
        except FujiError as exc:
            rest = tuple(c.name for c in writes[index + 1 :])
            return ApplyReport(diff, tuple(completed), change.name, exc, rest)
        completed.append(result)
    return ApplyReport(diff, tuple(completed))

async def calibration_status(self, *, timeout: float | None = None) -> CalibrationStatus:
    """What is calibrating, held or failed. Two transactions, the poll's status blocks."""
    status = await self._read_status("calibration_status", timeout)
    return operations.calibration_status(status)

async def channel_status(
    self, channel: ChannelId | str, *, timeout: float | None = None
) -> ChannelStatus:
    """One measured channel's status: range, calibration, hold and errors. Two transactions.

    Raises:
        FujiValidationError: ``channel`` is not one of channels 1-5; nothing was sent.
    """
    cid = coerce_channel(channel)
    if not cid.is_measured:
        msg = f"{cid.value} is a derived channel and has no status registers"
        raise FujiValidationError(msg, context=ErrorContext(channel=cid.value))
    status = await self._read_status("channel_status", timeout)
    return status.channels[cid]

async def close(self) -> None:
    """Close the analyzer and the port it opened. Idempotent.

    Waits for an operation in progress to finish. A transport the caller
    passed to ``open_device`` is left open.
    """
    await self._session.close()

async def diff_settings(
    self,
    document: SettingsDocument | Mapping[str, object],
    *,
    any_analyzer: bool = False,
    timeout: float | None = None,
) -> SettingsDiff:
    """Compare a settings document (``fujilib-settings/1``) with the analyzer's settings.

    Read-only: the range tables and every setting are read (four
    transactions), and each setting of the document is found unchanged, to
    be written, or refused, with the reason (:mod:`fujilib.devices.settings`).
    A document from another analyzer is refused unless ``any_analyzer``.

    Raises:
        FujiValidationError: the document is not a settings document.
        FujiError: a transaction failed.
    """
    doc = (
        document
        if isinstance(document, SettingsDocument)
        else SettingsDocument.from_json(dict(document))
    )
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> SettingsDiff:
        info = session.info
        if info is None:
            identity = await session.profile.identify(client, probe=True, deadline=deadline)
            info = session.learn_identity(identity)
        ranges = session.learn_ranges(await reads.read_ranges(client, deadline=deadline))
        current = await reads.read_settings(client, ranges=ranges, deadline=deadline)
        return diff_settings(
            doc,
            current,
            registry=session.profile.registry,
            ranges=ranges,
            serial_number=info.serial_number,
            any_analyzer=any_analyzer,
        )

    return await session.run("diff_settings", body, timeout=timeout)

async def identify(
    self,
    *,
    channel_map: Mapping[ChannelId | str, Gas | str] | None = None,
    timeout: float | None = None,
) -> DeviceInfo:
    """Read the type code, serial number, ranges and readings, and probe the capabilities.

    Six transactions (design §4.3). ``channel_map`` replaces the asserted
    gas labels; an asserted label is the only one fit for calculation
    (design §2.9).

    Raises:
        FujiValidationError: ``channel_map`` names an unknown channel or gas.
        FujiProtocolUnsupportedError: the type code does not name a ZP model.
        FujiError: a transaction failed.
    """
    asserted = coerce_channel_map(channel_map) if channel_map is not None else None
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> DeviceInfo:
        identity = await session.profile.identify(client, probe=True, deadline=deadline)
        return session.learn_identity(identity, channel_map=asserted)

    return await session.run("identify", body, timeout=timeout)

def manual_calibration(
    self,
    plan: ManualCalibrationPlan,
    *,
    gas: CalibrationGas | Mapping[ChannelId | str, CalibrationGas],
    confirm: bool = False,
    rule: SteadinessRule | None = None,
    adc: bool = False,
    interval: float = 0.5,
    key_timeout: float = 2.0,
    run_timeout: float = 30.0,
    cleanup_timeout: float = 30.0,
) -> RemoteCalibration:
    """A manual zero or span of ``plan``, driven from the host with the panel's keys.

    Use it as an async context manager (design §6.5)::

        plan = await anz.plan_manual_calibration("CH3", "span")
        gas = CalibrationGas(20.95, "vol%", label="20.95 % O2 in N2")
        async with anz.manual_calibration(plan, gas=gas, confirm=True) as run:
            await run.wait_steady()  # the operator opens the span-gas valve
            event = await run.calibrate(confirm=True)

    Entering it checks, before any key, that the panel is on the
    measurement screen with no calibration flag set, key lock and output
    hold are off, the analyzer reports no instrument error, the plan reads
    the same again and calibrates no channel on both ranges, and ``gas``
    (one for every established channel of the plan, or one for all) is
    each channel's calibration-gas setting. It then presses ZERO or SPAN,
    moves the cursor and selects the channel: the wait step, where the
    gas settles. These keys are ``STATEFUL``; ``calibrate(confirm=True)``
    sends the ENT that calibrates, which is ``DANGEROUS``. Leaving the
    block returns the panel to measurement for the step it is on (ESC on
    the wait step), shielded from cancellation.

    With ``adc`` the raw A/D values are read with each read on the wait
    step and kept in the record. ``interval`` is the time between those
    reads; the port is free between them, so a recording goes on.

    Raises:
        FujiValidationError: a gas is missing or malformed, or a time is
            not positive; nothing was sent.
    """
    return RemoteCalibration(
        self._session,
        plan,
        gas=gas,
        confirm=confirm,
        rule=rule,
        adc=adc,
        interval=interval,
        key_timeout=key_timeout,
        run_timeout=run_timeout,
        cleanup_timeout=cleanup_timeout,
    )

async def plan_auto_calibration(self, *, timeout: float | None = None) -> CalibrationPlan:
    """What :meth:`start_auto_calibration` would calibrate, against which gases, for how long.

    Read-only: the channels enabled for it, their ranges (both where the
    calibration range is "both"), the calibration gases, whether the
    outputs are held, and an estimated duration inferred from the flow
    times. Two transactions, plus the range tables if needed.
    """
    return await self._plan("plan_auto_calibration", CalibrationRun.AUTO_CALIBRATION, timeout)

async def plan_auto_zero_calibration(self, *, timeout: float | None = None) -> CalibrationPlan:
    """What :meth:`start_auto_zero_calibration` would zero; as :meth:`plan_auto_calibration`."""
    return await self._plan("plan_auto_zero_calibration", CalibrationRun.AUTO_ZERO, timeout)

async def plan_manual_calibration(
    self,
    channel: ChannelId | str,
    kind: ManualCalibrationKind | str,
    *,
    timeout: float | None = None,
) -> ManualCalibrationPlan:
    """What a manual zero or span of ``channel`` at the front panel would calibrate.

    Read-only. A zero of a channel set to "at once" zeroes every channel
    so set, and a channel set to "both" is calibrated on both its ranges;
    the plan lists every channel and range with its calibration gas
    (design §6.5). Two transactions, plus the range tables if needed.

    Raises:
        FujiValidationError: ``channel`` is not 1-5, or ``kind`` is not
            ``"zero"`` or ``"span"``; nothing was sent.
        FujiDecodeError: a range setting does not read as a range.
    """
    cid = _measured(channel)
    what = _manual_kind(kind)
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> ManualCalibrationPlan:
        ranges = await session.ensure_ranges(client, deadline)
        settings = await reads.read_registers(
            client, panel.MANUAL_PLAN_SETTINGS, ranges=ranges, deadline=deadline
        )
        return panel.plan_manual_calibration(
            settings,
            what,
            cid,
            ranges=ranges,
            established=[c.channel for c in session.channels],
        )

    return await session.run("plan_manual_calibration", body, timeout=timeout)

async def poll(self, *, detail: bool = True, timeout: float | None = None) -> Frame:
    """Read every established channel and the analyzer's status: two transactions.

    Without ``detail`` only the concentrations are read (one transaction),
    and every reading's validity is unknown rather than assumed (design
    §4.3). A channel that reads non-zero for the first time joins the
    established channels, in this frame and every later one (design §2.9).

    Raises:
        FujiError: a transaction failed. When the status block fails after
            the concentrations were read, the error's ``extra["completed"]``
            holds the concentration words.
    """
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> Frame:
        return session.learn_poll(
            await reads.read_poll(client, detail=detail, deadline=deadline)
        )

    return await session.run("poll", body, timeout=timeout)

async def read_adc(self, *, timeout: float | None = None) -> AdcValues:
    """The 21 raw A/D counts of the service manual's table (undocumented; design §6.6).

    A service diagnostic, not a calibrated or higher-resolution gas measurement.

    Raises:
        FujiCapabilityError: the analyzer has no A/D block; nothing was sent once known.
    """
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> AdcValues:
        return await session.read_capability(
            Capability.ADC_VALUES, "read_adc", lambda: reads.read_adc(client, deadline=deadline)
        )

    return await session.run("read_adc", body, timeout=timeout, requires=Capability.ADC_VALUES)

async def read_calibration_log(
    self, channel: ChannelId | str | None = None, *, timeout: float | None = None
) -> tuple[CalibrationLogEntry, ...]:
    """Calibration-log records, newest first per channel; firmware 2.24 or later.

    ``channel`` ``None`` reads every established measured channel, in order.
    Each channel is seven transactions.

    Raises:
        FujiValidationError: ``channel`` is not one of channels 1-5; nothing was sent.
        FujiFirmwareError: the analyzer has no calibration log (older
            firmware); nothing was sent once known.
    """
    if channel is not None:
        wanted: tuple[ChannelId, ...] = (coerce_channel(channel),)
    else:
        wanted = tuple(c.channel for c in self._session.channels if c.channel.is_measured)
    for cid in wanted:
        if not cid.is_measured:
            msg = f"{cid.value} is a derived channel and has no calibration log"
            raise FujiValidationError(msg, context=ErrorContext(channel=cid.value))
    session = self._session

    async def body(
        client: ProtocolClient, deadline: Deadline
    ) -> tuple[CalibrationLogEntry, ...]:
        entries: list[CalibrationLogEntry] = []
        for cid in wanted:
            entries += await session.read_capability(
                Capability.CALIBRATION_LOG,
                "read_calibration_log",
                partial(reads.read_calibration_log, client, cid, deadline=deadline),
            )
        return tuple(entries)

    return await session.run(
        "read_calibration_log", body, timeout=timeout, requires=Capability.CALIBRATION_LOG
    )

async def read_channel(
    self, channel: ChannelId | str, *, timeout: float | None = None
) -> Reading:
    """One established channel's reading, from a full poll.

    Raises:
        FujiValidationError: ``channel`` is unknown or not established;
            nothing was sent. Assert it with ``channel_map`` to read it.
    """
    cid = self._established(channel)
    frame = await self.poll(timeout=timeout)
    return frame.channel(cid)

async def read_clock(self, *, timeout: float | None = None) -> ClockReading:
    """The analyzer's real-time clock and when it was read (undocumented; design §6.6).

    Naive local time with a two-digit year; it drifts, and it never replaces
    host timestamps.

    Raises:
        FujiCapabilityError: the analyzer has no clock; nothing was sent once known.
        FujiDecodeError: the words are not a date.
    """
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> ClockReading:
        return await session.read_capability(
            Capability.CLOCK, "read_clock", lambda: reads.read_clock(client, deadline=deadline)
        )

    return await session.run("read_clock", body, timeout=timeout, requires=Capability.CLOCK)

async def read_error_log(self, *, timeout: float | None = None) -> tuple[ErrorLogEntry, ...]:
    """The error log, newest first; up to 14 entries with day, hour and minute only."""

    async def body(client: ProtocolClient, deadline: Deadline) -> tuple[ErrorLogEntry, ...]:
        return await reads.read_error_log(client, deadline=deadline)

    return await self._session.run("read_error_log", body, timeout=timeout)

async def read_metadata(self, *, timeout: float | None = None) -> AnalyzerMetadata:
    """The settings snapshot a consumer records with its data (design §2.11, §7.2).

    Response times, averaging, calibration gases and scope, hold, the
    automatic schedules, the current ranges and the analyzer's clock. Reads
    the identity first if ``identify()`` has not run, and the range tables
    if a poll saw a range change.

    Raises:
        FujiError: a transaction failed.
    """
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> AnalyzerMetadata:
        info = session.info
        if info is None:
            identity = await session.profile.identify(client, probe=True, deadline=deadline)
            info = session.learn_identity(identity)
        ranges = await session.ensure_ranges(client, deadline)
        clock = await self._clock_available(client, deadline)
        meta = await reads.read_metadata(
            client,
            serial_number=info.serial_number,
            ranges=ranges,
            channels=session.channels,
            clock=clock,
            deadline=deadline,
        )
        session.learn_current_ranges(meta.current_range)
        return meta

    return await session.run("read_metadata", body, timeout=timeout)

async def read_parameter(
    self,
    name: str,
    *,
    alarm_targets: Mapping[int, ChannelId | str] | None = None,
    timeout: float | None = None,
) -> RegisterValue:
    """One register by its name in the register map (``docs/registers.md``).

    A range-scaled value uses the range tables, read first if needed. An
    alarm limit needs ``alarm_targets`` (alarm number to channel), because
    the target register's encoding is contested (design §5.2); without it
    the value is ``None`` and only the raw word is given.

    Raises:
        FujiValidationError: ``name`` is not in the register map; nothing was sent.
    """
    values = await self._read_named("read_parameter", (name,), alarm_targets, timeout)
    return values[name]

async def read_parameters(
    self,
    names: Iterable[str],
    *,
    alarm_targets: Mapping[int, ChannelId | str] | None = None,
    timeout: float | None = None,
) -> Mapping[str, RegisterValue]:
    """Several registers by name, read in the fewest blocks, in the order given.

    Raises:
        FujiValidationError: a name is not in the register map; nothing was sent.
    """
    if isinstance(names, str):
        msg = "names must be a sequence of register names, not one string"
        raise FujiValidationError(msg)
    return await self._read_named("read_parameters", tuple(names), alarm_targets, timeout)

async def read_ranges(self, *, timeout: float | None = None) -> tuple[RangeInfo, ...]:
    """Read the range tables of channels 1-5; the session keeps them. One transaction."""
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> tuple[RangeInfo, ...]:
        return session.learn_ranges(await reads.read_ranges(client, deadline=deadline))

    return await session.run("read_ranges", body, timeout=timeout)

async def read_settings(
    self,
    *,
    alarm_targets: Mapping[int, ChannelId | str] | None = None,
    timeout: float | None = None,
) -> Mapping[str, RegisterValue]:
    """Every holding register, decoded, by name. Three transactions, plus the ranges if needed.

    Raises:
        FujiError: a transaction failed.
    """
    targets = _alarm_targets(alarm_targets)
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> Mapping[str, RegisterValue]:
        ranges = await session.ensure_ranges(client, deadline)
        return await reads.read_settings(
            client, ranges=ranges, alarm_targets=targets, deadline=deadline
        )

    return await session.run("read_settings", body, timeout=timeout)

async def reopen(self, *, timeout: float | None = None) -> DeviceInfo:
    """Open the port again and identify the analyzer, after a connection failure.

    A connection failure breaks the session (every later call is refused);
    this is the way back without losing what the session learned. Only a
    port that ``open_device`` opened by name can be reopened. The station
    must answer as the same analyzer: the same serial number and type code.

    Raises:
        FujiConfigurationError: the port came from the caller; or another
            analyzer answers on the station.
        FujiConnectionError: the analyzer is closed, or the port cannot be opened.
        FujiError: identification failed; the analyzer stays unusable.
    """
    return await self._session.reopen(timeout=timeout)

async def reprobe(
    self, capability: Capability, *, timeout: float | None = None
) -> Availability:
    """Probe ``capability`` again and return what was found (design §6.6).

    Raises:
        FujiValidationError: ``capability`` is not a probed capability; nothing was sent.
    """
    if capability not in PROBED_CAPABILITIES:
        names = ", ".join(str(c.name) for c in PROBED_CAPABILITIES)
        msg = f"{capability!r} is not a probed capability; probed are {names}"
        raise FujiValidationError(msg)
    session = self._session

    async def body(client: ProtocolClient, deadline: Deadline) -> Availability:
        probe = await reads.probe_capability(client, capability, deadline=deadline)
        session.set_availability(capability, probe.availability)
        return probe.availability

    return await session.run("reprobe", body, timeout=timeout)

async def return_to_measurement(
    self, *, confirm: bool = False, timeout: float | None = None
) -> CommandResult:
    """Put the front panel back on the measurement screen (42002). STATEFUL.

    It takes an operator out of whatever menu they are in, and out of a
    manual calibration's channel selection, so it is not refused there. It
    does not stop or cancel a calibration, and is refused while one is under
    way: on a manual calibration's wait step 42002 brings the display back
    but leaves the calibration's flag set, and nothing at the panel shows it
    (protocol findings §18.4, design §13.1 #83). ESC on the wait step, at
    the panel, cancels a manual calibration.

    Raises:
        FujiConfirmationRequiredError: ``confirm`` is not ``True``; nothing was sent.
        FujiAnalyzerStateError: a calibration, automatic or manual, is under
            way; nothing was sent.
        FujiVerificationError: acknowledged, but the panel does not show
            the measurement screen; or it shows it with a calibration flag
            set, which a calibration started at the panel meanwhile would do.
        FujiWriteOutcomeUnknownError: its reply was lost and the status
            cannot be read.
        FujiError: a transaction failed.
    """
    return await self._command("return_to_measurement", None, confirm, timeout)

async def set_calibration_gas(
    self,
    channel: ChannelId | str,
    range_number: int,
    kind: str,
    value: float | str,
    *,
    unit: Unit | str,
    confirm: bool = False,
    timeout: float | None = None,
) -> WriteResult:
    """Set the zero or span calibration gas of a measured channel's range. DANGEROUS.

    It takes effect at the next calibration, manual or automatic, and a
    wrong value miscalibrates the analyzer then. ``unit`` must be the
    range's own; span gas is limited to 1-105 % and zero gas to 0-100 % of
    the range's full scale.

    Raises:
        FujiValidationError: ``channel``, ``range_number`` or ``kind``
            (``"zero"`` or ``"span"``) is not one there is, or the value
            does not fit the range.
        FujiError: as :meth:`write_parameter`.
    """
    cid = _measured(channel)
    if (
        range_number not in {1, 2}
        or isinstance(range_number, bool)
        or kind
        not in {
            "zero",
            "span",
        }
    ):
        msg = f"expected range 1 or 2 and kind 'zero' or 'span', got {range_number!r}, {kind!r}"
        raise FujiValidationError(msg, context=ErrorContext(channel=cid.value))
    spec = self._writable(f"calibration_gas.ch{cid.number}.range{range_number}.{kind}")
    return await self._write(
        "set_calibration_gas", spec, value, unit=unit, confirm=confirm, timeout=timeout
    )

async def set_hold_mode(
    self, mode: HoldMode | str, *, confirm: bool = False, timeout: float | None = None
) -> WriteResult:
    """What the outputs hold during calibration: ``last_value`` or ``setting``.

    Raises:
        FujiError: as :meth:`write_parameter`.
    """
    spec = self._writable("hold.mode")
    return await self._write("set_hold_mode", spec, mode, confirm=confirm, timeout=timeout)

async def set_hold_value(
    self,
    channel: ChannelId | str,
    percent_fs: int,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> WriteResult:
    """The value a measured channel holds in ``setting`` mode, 0-100 % of full scale.

    Raises:
        FujiValidationError: ``channel`` is not one of channels 1-5.
        FujiError: as :meth:`write_parameter`.
    """
    cid = _measured(channel)
    spec = self._writable(f"hold.ch{cid.number}.value")
    return await self._write(
        "set_hold_value", spec, percent_fs, confirm=confirm, timeout=timeout
    )

async def set_output_hold(
    self, enabled: bool, *, confirm: bool = False, timeout: float | None = None
) -> WriteResult:
    """Hold the outputs, and the Modbus concentrations, during calibration, or not.

    Raises:
        FujiError: as :meth:`write_parameter`.
    """
    spec = self._writable("output_hold.enabled")
    return await self._write("set_output_hold", spec, enabled, confirm=confirm, timeout=timeout)

async def set_range(
    self,
    channel: ChannelId | str,
    range_number: int,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> WriteResult:
    """Select range 1 or 2 of a measured channel, whose range method must be manual.

    It returns once the channel measures on the range: the analyzer
    switches some tens of milliseconds after the setting reads back, so
    the channel's current range is read until it follows, within the
    read-back budget.

    Raises:
        FujiValidationError: ``channel`` is not one of channels 1-5,
            ``range_number`` is not 1 or 2, or the channel's range method
            is not manual; nothing was written.
        FujiVerificationError: the setting reads back otherwise, or it
            reads back as written but the channel did not switch to it.
        FujiError: as :meth:`write_parameter`.
    """
    cid = _measured(channel)
    if range_number not in {1, 2} or isinstance(range_number, bool):
        msg = f"range_number must be 1 or 2, got {range_number!r}"
        raise FujiValidationError(msg, context=ErrorContext(channel=cid.value))
    spec = self._writable(f"range.ch{cid.number}.selected")
    index = RangeIndex(range_number - 1)
    return await self._write("set_range", spec, index, confirm=confirm, timeout=timeout)