servomexlib.sync¶

def analyser_status(self, *, timeout: float | None = None) -> AnalyserStatus:
    """Return the analyser-level status."""
    return self._portal.call(self._analyzer.analyser_status, timeout=timeout)

def calibration_status(
    self, group: int = 1, *, timeout: float | None = None
) -> CalibrationProgress:
    """Return autocalibration progress for ``group`` (Modbus only)."""
    return self._portal.call(self._analyzer.calibration_status, group, timeout=timeout)

def identify(self, *, timeout: float | None = None) -> DeviceInfo:
    """Return device identity, caching it."""
    return self._portal.call(self._analyzer.identify, timeout=timeout)

@classmethod
def open(
    cls,
    port: str | Transport,
    *,
    protocol: ProtocolKind | str = ProtocolKind.AUTO,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
    identify: bool = True,
    backend: str = "asyncio",
    portal: SyncPortal | None = None,
) -> SyncAnalyzer:
    """Open an analyser synchronously and return an entered :class:`SyncAnalyzer`.

    Pass ``portal`` to share an existing loop (e.g. one already hosting an
    in-process fake); otherwise a private background loop is started and torn
    down with the analyzer.
    """
    resolved = ProtocolKind(protocol)  # accepts a ProtocolKind or its str value
    owns_portal = portal is None
    active = portal if portal is not None else SyncPortal(backend=backend)
    if owns_portal:
        active.__enter__()
    try:
        analyzer = active.call(
            open_device,
            port,
            protocol=resolved,
            address=address,
            serial_settings=serial_settings,
            timeout=timeout,
            identify=identify,
        )
        # Enter via the portal's wrapper so __aenter__/__aexit__ share one task.
        managed = active.wrap_async_context_manager(analyzer)
        managed.__enter__()
    except BaseException:
        if owns_portal:
            active.__exit__(None, None, None)
        raise
    return cls(active, analyzer, managed, owns_portal=owns_portal)

def poll(self, *, wait_fresh: bool = False, timeout: float | None = None) -> Frame:
    """Return one frame (all channels + analyser status)."""
    return self._portal.call(self._analyzer.poll, wait_fresh=wait_fresh, timeout=timeout)

def read_all(self, *, timeout: float | None = None) -> dict[ChannelId, Reading]:
    """Return every channel's latest reading keyed by id."""
    return self._portal.call(self._analyzer.read_all, timeout=timeout)

def read_channel(self, channel: ChannelId | str, *, timeout: float | None = None) -> Reading:
    """Return one channel's latest reading."""
    return self._portal.call(self._analyzer.read_channel, channel, timeout=timeout)

def snapshot(self) -> Frame:
    """Return the cached latest frame without any I/O."""
    return self._analyzer.snapshot()

def start_calibration(
    self, group: int, *, confirm: bool = False, timeout: float | None = None
) -> None:
    """Start autocalibration for ``group`` (Modbus only; requires ``confirm=True``)."""
    self._portal.call(self._analyzer.start_calibration, group, confirm=confirm, timeout=timeout)

def status(self, channel: ChannelId | str, *, timeout: float | None = None) -> ChannelStatus:
    """Return one channel's latest status."""
    return self._portal.call(self._analyzer.status, channel, timeout=timeout)

def stop_calibration(self, *, confirm: bool = False, timeout: float | None = None) -> None:
    """Stop all autocalibration (Modbus only; requires ``confirm=True``)."""
    self._portal.call(self._analyzer.stop_calibration, confirm=confirm, timeout=timeout)

def stream(
    self, *, mode: StreamMode | None = None, rate_hz: float | None = None
) -> SyncStreamingSession:
    """Return a blocking iterator over the analyser's sample stream."""
    session = self._portal.call(_make_stream, self._analyzer, mode, rate_hz)
    return SyncStreamingSession(self._portal, session)

def add(
    self,
    name: str,
    source: Analyzer | str | Transport,
    *,
    protocol: ProtocolKind | str = ProtocolKind.MODBUS_RTU,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
) -> Analyzer:
    """Register an analyser; returns the underlying async :class:`Analyzer`."""
    return self._portal.call(
        self._manager.add,
        name,
        source,
        protocol=ProtocolKind(protocol),
        address=address,
        serial_settings=serial_settings,
        timeout=timeout,
    )

def get(self, name: str) -> Analyzer:
    """Return the analyser registered under ``name`` (no I/O)."""
    return self._manager.get(name)

def poll(
    self, names: Sequence[str] | None = None, *, timeout: float | None = None
) -> Mapping[str, DeviceResult[Frame]]:
    """Read one :class:`Frame` per (or named) analyser, keyed by name."""
    return self._portal.call(self._manager.poll, names, timeout=timeout)

def poll_samples(
    self, *, names: Sequence[str] | None = None, timeout: float | None = None
) -> list[Sample]:
    """Poll every (or named) analyser → flat samples."""
    return self._portal.call(self._manager.poll_samples, names=names, timeout=timeout)

def remove(self, name: str) -> None:
    """Unregister ``name``, closing the shared transport on the last ref."""
    self._portal.call(self._manager.remove, name)

def call(self, func: Callable[..., Awaitable[T]], /, *args: object, **kwargs: object) -> T:
    """Run ``func(*args, **kwargs)`` on the loop thread and return its result.

    The portal re-raises the concrete exception the coroutine raised, so sync
    callers catch the same :class:`ServomexError` subclasses as async ones.
    """
    target = functools.partial(func, *args, **kwargs)
    return self._require_portal().call(target)

def wrap_async_context_manager(
    self, cm: AbstractAsyncContextManager[T]
) -> AbstractContextManager[T]:
    """Wrap an async context manager so its enter/exit run in one portal task.

    ``call(cm.__aenter__)`` then ``call(cm.__aexit__)`` would run in *different*
    loop tasks and trip anyio's cancel-scope affinity check; this bridges them.
    """
    return self._require_portal().wrap_async_context_manager(cm)

def close(self) -> None:
    """Close the backing sink."""
    self._portal.call(self._sink.close)

def open(self) -> None:
    """Open the backing sink."""
    self._portal.call(self._sink.open)

def write_many(self, samples: Sequence[Sample]) -> None:
    """Append ``samples`` to the backing sink."""
    self._portal.call(self._sink.write_many, samples)