alicatlib.sync¶

@staticmethod
@contextmanager
def open(
    port: str | Transport | AlicatProtocolClient,
    *,
    unit_id: str = "A",
    serial: SerialSettings | None = None,
    timeout: float = 0.5,
    recover_from_stream: bool = True,
    model_hint: str | None = None,
    assume_capabilities: Capability = Capability.NONE,
    assume_media: Medium | None = None,
    portal: SyncPortal | None = None,
) -> Generator[SyncDevice]:
    """Open a sync :class:`SyncDevice` scoped to a ``with`` block.

    Mirrors :func:`alicatlib.devices.factory.open_device` parameter
    for parameter. The returned sync CM drives the async factory
    through a :class:`SyncPortal`; the portal is created per-call
    unless a pre-existing one is passed in via ``portal=``.

    Passing a :class:`Transport` or
    :class:`AlicatProtocolClient` is advanced: the caller is
    responsible for ensuring the object was constructed inside the
    portal's event loop (or is loop-agnostic). The common case —
    passing a ``str`` port path — creates the transport inside the
    portal and avoids that concern.
    """
    with ExitStack() as stack:
        active_portal = portal if portal is not None else stack.enter_context(SyncPortal())
        async_cm = open_device(
            port,
            unit_id=unit_id,
            serial=serial,
            timeout=timeout,
            recover_from_stream=recover_from_stream,
            model_hint=model_hint,
            assume_capabilities=assume_capabilities,
            assume_media=assume_media,
        )
        async_device = stack.enter_context(active_portal.wrap_async_context_manager(async_cm))
        yield wrap_device(async_device, active_portal)

def target(self) -> str:
    """Return a log-safe description of the target: ``host:port/db.schema.table``."""
    if self.dsn is not None:
        parsed = urlparse(self.dsn)
        host = parsed.hostname or "?"
        port = parsed.port or self.port
        db = (parsed.path or "/?").lstrip("/") or "?"
    else:
        host = self.host or "?"
        port = self.port
        db = self.database or "?"
    return f"{host}:{port}/{db}.{self.schema}.{self.table}"

def __enter__(self) -> Self:
    """Start the portal, build the async manager, enter its CM."""
    if self._entered:
        raise RuntimeError("SyncAlicatManager is not reusable after exit")
    self._entered = True
    stack = ExitStack()
    try:
        portal = (
            self._portal_override
            if self._portal_override is not None
            else stack.enter_context(SyncPortal())
        )
        mgr = AlicatManager(error_policy=self._error_policy)
        stack.enter_context(portal.wrap_async_context_manager(mgr))
        self._portal = portal
        self._mgr = mgr
        self._stack = stack
    except BaseException:
        stack.close()
        self._portal = None
        self._mgr = None
        raise
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close the managed devices + portal (if owned)."""
    stack, self._stack = self._stack, None
    self._wrapped.clear()
    self._mgr = None
    self._portal = None
    if stack is not None:
        stack.__exit__(exc_type, exc, tb)

def add(
    self,
    name: str,
    source: SyncDevice | Device | str | Transport | AlicatProtocolClient,
    *,
    unit_id: str = "A",
    serial: SerialSettings | None = None,
    timeout: float = 0.5,
) -> SyncDevice:
    """Blocking :meth:`AlicatManager.add`.

    Accepts a :class:`SyncDevice` as ``source`` in addition to the
    async shapes — the wrapper is unwrapped to the underlying
    :class:`Device` before delegation, matching the manager's
    "pre-built device, no lifecycle ownership" contract.
    """
    mgr = self._require_mgr()
    async_source: Device | str | Transport | AlicatProtocolClient = unwrap_sync_device(source)
    async_device = self.portal.call(
        mgr.add,
        name,
        async_source,
        unit_id=unit_id,
        serial=serial,
        timeout=timeout,
    )
    wrapped = wrap_device(async_device, self.portal)
    self._wrapped[name] = wrapped
    return wrapped

def close(self) -> None:
    """Blocking :meth:`AlicatManager.close` — idempotent."""
    self._wrapped.clear()
    mgr = self._mgr
    if mgr is None:
        return
    portal = self._portal
    if portal is None:
        return
    portal.call(mgr.close)

def execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    requests_by_name: Mapping[str, Req],
) -> Mapping[str, DeviceResult[Resp]]:
    """Blocking :meth:`AlicatManager.execute`."""
    mgr = self._require_mgr()
    return self.portal.call(mgr.execute, command, requests_by_name)

def get(self, name: str) -> SyncDevice:
    """Return the sync wrapper for the device registered under ``name``.

    The async manager's :meth:`AlicatManager.get` is already
    synchronous — this method caches the sync wrapper so repeated
    ``get`` calls return the same :class:`SyncDevice` instance per
    device name.
    """
    cached = self._wrapped.get(name)
    if cached is not None:
        return cached
    mgr = self._require_mgr()
    async_device = mgr.get(name)  # raises AlicatValidationError on unknown
    wrapped = wrap_device(async_device, self.portal)
    self._wrapped[name] = wrapped
    return wrapped

def poll(
    self,
    names: Sequence[str] | None = None,
) -> Mapping[str, DeviceResult[DataFrame]]:
    """Blocking :meth:`AlicatManager.poll`."""
    mgr = self._require_mgr()
    return self.portal.call(mgr.poll, names)

def remove(self, name: str) -> None:
    """Blocking :meth:`AlicatManager.remove`."""
    mgr = self._require_mgr()
    self._wrapped.pop(name, None)
    self.portal.call(mgr.remove, name)

def request(
    self,
    statistics: Sequence[Statistic | str],
    names: Sequence[str] | None = None,
    *,
    averaging_ms: int = 1,
) -> Mapping[str, DeviceResult[MeasurementSet]]:
    """Blocking :meth:`AlicatManager.request`."""
    mgr = self._require_mgr()
    return self.portal.call(
        mgr.request,
        statistics,
        names,
        averaging_ms=averaging_ms,
    )

def __enter__(self) -> Self:
    """Nesting convenience — matches :meth:`Device.__aenter__`."""
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close the device on exit — session close is idempotent."""
    del exc_type, exc, tb
    self.close()

def analog_output_source(
    self,
    channel: AnalogOutputChannel = _ANALOG_OUTPUT_PRIMARY,
    value: int | None = None,
    unit_code: int | None = None,
) -> AnalogOutputSourceSetting:
    """Blocking :meth:`Device.analog_output_source`.

    Default channel is :attr:`AnalogOutputChannel.PRIMARY` — same
    default as the async side, bound at module load via the
    module-level alias so sync-parity signature comparison sees
    the same sentinel object.
    """
    return self._portal.call(self._dev.analog_output_source, channel, value, unit_code)

def average_timing(
    self,
    statistic_code: int,
    averaging_ms: int | None = None,
) -> AverageTimingSetting:
    """Blocking :meth:`Device.average_timing`."""
    return self._portal.call(self._dev.average_timing, statistic_code, averaging_ms)

def blink_display(self, duration_s: int | None = None) -> BlinkDisplayState:
    """Blocking :meth:`Device.blink_display`."""
    return self._portal.call(self._dev.blink_display, duration_s)

def close(self) -> None:
    """Release the underlying session — idempotent.

    Prefer ``with Alicat.open(...) as dev:`` over calling this by
    hand; the outer context manager owns transport lifecycle.
    """
    self._portal.call(self._dev.close)

def engineering_units(
    self,
    statistic: Statistic | str,
    unit: Unit | int | str | None = None,
    *,
    apply_to_group: bool = False,
    override_special_rules: bool = False,
) -> UnitSetting:
    """Blocking :meth:`Device.engineering_units`."""
    return self._portal.call(
        self._dev.engineering_units,
        statistic,
        unit,
        apply_to_group=apply_to_group,
        override_special_rules=override_special_rules,
    )

def execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    request: Req,
) -> Resp:
    """Blocking :meth:`Device.execute`."""
    return self._portal.call(self._dev.execute, command, request)

def full_scale(self, statistic: Statistic | str) -> FullScaleValue:
    """Blocking :meth:`Device.full_scale`."""
    return self._portal.call(self._dev.full_scale, statistic)

def gas(
    self,
    gas: Gas | str | None = None,
    *,
    save: bool | None = None,
) -> GasState:
    """Blocking :meth:`Device.gas`."""
    return self._portal.call(self._dev.gas, gas, save=save)

def gas_list(self) -> Mapping[int, str]:
    """Blocking :meth:`Device.gas_list`."""
    return self._portal.call(self._dev.gas_list)

def lock_display(self) -> DisplayLockResult:
    """Blocking :meth:`Device.lock_display`."""
    return self._portal.call(self._dev.lock_display)

def poll(self) -> DataFrame:
    """Blocking :meth:`Device.poll`."""
    return self._portal.call(self._dev.poll)

def power_up_tare(self, enable: bool | None = None) -> PowerUpTareState:
    """Blocking :meth:`Device.power_up_tare`."""
    return self._portal.call(self._dev.power_up_tare, enable)

def request(
    self,
    statistics: Sequence[Statistic | str],
    *,
    averaging_ms: int = 1,
) -> MeasurementSet:
    """Blocking :meth:`Device.request`."""
    return self._portal.call(
        self._dev.request,
        statistics,
        averaging_ms=averaging_ms,
    )

def stp_ntp_pressure(
    self,
    mode: StpNtpMode,
    pressure: float | None = None,
    unit_code: int | None = None,
) -> StpNtpPressureSetting:
    """Blocking :meth:`Device.stp_ntp_pressure`."""
    return self._portal.call(self._dev.stp_ntp_pressure, mode, pressure, unit_code)

def stp_ntp_temperature(
    self,
    mode: StpNtpMode,
    temperature: float | None = None,
    unit_code: int | None = None,
) -> StpNtpTemperatureSetting:
    """Blocking :meth:`Device.stp_ntp_temperature`."""
    return self._portal.call(self._dev.stp_ntp_temperature, mode, temperature, unit_code)

def stream(
    self,
    *,
    rate_ms: int | None = None,
    strict: bool = False,
    overflow: OverflowPolicy | None = None,
    buffer_size: int = 256,
) -> SyncStreamingSession:
    """Blocking :meth:`Device.stream` — returns a sync context manager.

    The returned :class:`SyncStreamingSession` is both a sync
    context manager and a sync iterator; use it as::

        with sync_dev.stream(rate_ms=50) as stream:
            for frame in stream:
                process(frame)
    """
    async_stream = self._dev.stream(
        rate_ms=rate_ms,
        strict=strict,
        overflow=overflow,
        buffer_size=buffer_size,
    )
    return SyncStreamingSession(async_stream, self._portal)

def tare_absolute_pressure(self) -> TareResult:
    """Blocking :meth:`Device.tare_absolute_pressure`."""
    return self._portal.call(self._dev.tare_absolute_pressure)

def tare_flow(self) -> TareResult:
    """Blocking :meth:`Device.tare_flow`."""
    return self._portal.call(self._dev.tare_flow)

def tare_gauge_pressure(self) -> TareResult:
    """Blocking :meth:`Device.tare_gauge_pressure`."""
    return self._portal.call(self._dev.tare_gauge_pressure)

def totalizer_config(
    self,
    totalizer: TotalizerId = _TOTALIZER_FIRST,
    *,
    flow_statistic_code: int | None = None,
    mode: TotalizerMode | None = None,
    limit_mode: TotalizerLimitMode | None = None,
    digits: int | None = None,
    decimal_place: int | None = None,
) -> TotalizerConfig:
    """Blocking :meth:`Device.totalizer_config`."""
    return self._portal.call(
        self._dev.totalizer_config,
        totalizer,
        flow_statistic_code=flow_statistic_code,
        mode=mode,
        limit_mode=limit_mode,
        digits=digits,
        decimal_place=decimal_place,
    )

def totalizer_reset(
    self,
    totalizer: TotalizerId = _TOTALIZER_FIRST,
    *,
    confirm: bool = False,
) -> TotalizerResetResult:
    """Blocking :meth:`Device.totalizer_reset` — destructive; requires ``confirm=True``."""
    return self._portal.call(self._dev.totalizer_reset, totalizer, confirm=confirm)

def totalizer_reset_peak(
    self,
    totalizer: TotalizerId = _TOTALIZER_FIRST,
    *,
    confirm: bool = False,
) -> TotalizerResetResult:
    """Blocking :meth:`Device.totalizer_reset_peak` — destructive."""
    return self._portal.call(self._dev.totalizer_reset_peak, totalizer, confirm=confirm)

def totalizer_save(
    self,
    enable: bool | None = None,
    *,
    save: bool | None = None,
) -> TotalizerSaveState:
    """Blocking :meth:`Device.totalizer_save`."""
    return self._portal.call(self._dev.totalizer_save, enable, save=save)

def unlock_display(self) -> DisplayLockResult:
    """Blocking :meth:`Device.unlock_display`."""
    return self._portal.call(self._dev.unlock_display)

def user_data(self, slot: int, value: str | None = None) -> UserDataSetting:
    """Blocking :meth:`Device.user_data`."""
    return self._portal.call(self._dev.user_data, slot, value)

def zero_band(self, zero_band: float | None = None) -> ZeroBandSetting:
    """Blocking :meth:`Device.zero_band`."""
    return self._portal.call(self._dev.zero_band, zero_band)

def __enter__(self) -> Self:
    """Start the portal's background thread and event loop."""
    if self._entered:
        raise RuntimeError("SyncPortal is not reusable after exit")
    self._entered = True
    cm = start_blocking_portal(self._backend)
    self._portal = cm.__enter__()
    self._cm = cm
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Stop the portal and join its thread."""
    cm, self._cm = self._cm, None
    self._portal = None
    if cm is not None:
        cm.__exit__(exc_type, exc, tb)

def call[**P, T](
    self,
    func: Callable[P, Awaitable[T]],
    *args: P.args,
    **kwargs: P.kwargs,
) -> T:
    """Run ``func(*args, **kwargs)`` on the portal's event loop.

    Single-member :class:`ExceptionGroup` wrappers are stripped so
    callers can catch concrete exception types.
    """
    portal = self._require_portal()
    bound: Callable[[], Awaitable[T]] = (
        partial(func, *args, **kwargs) if kwargs else partial(func, *args)
    )
    try:
        return portal.call(bound)
    except Exception as exc:
        unwrapped = _unwrap_single_group(exc)
        if unwrapped is exc:
            raise
        raise unwrapped from None

def wrap_async_context_manager[T](
    self, acm: AbstractAsyncContextManager[T]
) -> AbstractContextManager[T]:
    """Present an async context manager as a sync context manager."""
    return self._require_portal().wrap_async_context_manager(acm)

def wrap_async_iter[T](self, async_iter: AsyncIterator[T]) -> SyncAsyncIterator[T]:
    """Present an async iterator as a blocking, closeable iterator.

    The returned object is iterable (``for x in it: ...``) and
    supports :meth:`close` / context-manager use so outer wrappers
    can cancel the producer on early exit.
    """
    self._require_portal()
    return SyncAsyncIterator(self, async_iter)

def __enter__(self) -> Self:
    """Start the portal, open the async sink."""
    if self._entered:
        raise RuntimeError("SyncSinkAdapter is not reusable after exit")
    self._entered = True
    stack = ExitStack()
    try:
        self._portal = (
            self._portal_override
            if self._portal_override is not None
            else stack.enter_context(SyncPortal())
        )
        self.open()
        self._stack = stack
    except BaseException:
        stack.close()
        self._portal = None
        raise
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close the sink and, if owned, stop the portal."""
    stack, self._stack = self._stack, None
    try:
        self.close()
    finally:
        self._portal = None
        if stack is not None:
            stack.__exit__(exc_type, exc, tb)