sartoriuslib.sync¶

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}"

@staticmethod
@contextmanager
def open(
    port: str | Transport,
    *,
    protocol: ProtocolKind | None = None,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
    src_sbn: int = 0x01,
    dst_sbn: int = 0x09,
    strict: bool = False,
    identify: bool = True,
    portal: SyncPortal | None = None,
) -> Generator[SyncBalance]:
    """Open a sync :class:`SyncBalance` scoped to a ``with`` block.

    Mirrors :func:`sartoriuslib.open_device` parameter-for-
    parameter (modulo the portal plumbing). The sync CM drives
    the async factory through a :class:`SyncPortal`; the portal
    is created per-call unless one is passed in via ``portal=``.
    """
    # Local import keeps the ProtocolKind value available at runtime
    # (the top-level import is guarded by TYPE_CHECKING).
    from sartoriuslib.protocol.base import ProtocolKind as _ProtocolKind  # noqa: PLC0415

    effective_protocol = protocol if protocol is not None else _ProtocolKind.XBPI

    with ExitStack() as stack:
        active_portal = portal if portal is not None else stack.enter_context(SyncPortal())
        balance = active_portal.call(
            open_device,
            port,
            protocol=effective_protocol,
            serial_settings=serial_settings,
            timeout=timeout,
            src_sbn=src_sbn,
            dst_sbn=dst_sbn,
            strict=strict,
            identify=identify,
        )
        try:
            yield wrap_balance(balance, active_portal)
        finally:
            # Close the underlying transport through the portal;
            # the Balance's aclose closes the transport it was
            # constructed against.
            active_portal.call(balance.aclose)

def capacity(self, area: int = 0) -> Quantity:
    """Blocking :meth:`Balance.capacity`."""
    return self._portal.call(self._bal.capacity, area)

def configure_protocol(
    self,
    target: ProtocolKind,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
    timeout: float | None = None,
    confirm: bool = False,
) -> DeviceInfo:
    """Blocking :meth:`Balance.configure_protocol`."""
    return self._portal.call(
        self._bal.configure_protocol,
        target,
        baudrate=baudrate,
        parity=parity,
        stopbits=stopbits,
        timeout=timeout,
        confirm=confirm,
    )

def discover_temperature_sensors(self, *, max_index: int = 8) -> tuple[int, ...]:
    """Blocking :meth:`Balance.discover_temperature_sensors`."""
    return self._portal.call(
        self._bal.discover_temperature_sensors,
        max_index=max_index,
    )

def get_auto_zero(self) -> AutoZeroMode:
    """Blocking :meth:`Balance.get_auto_zero`."""
    return self._portal.call(self._bal.get_auto_zero)

def get_display_unit(self) -> Unit:
    """Blocking :meth:`Balance.get_display_unit`."""
    return self._portal.call(self._bal.get_display_unit)

def get_filter_mode(self) -> FilterMode:
    """Blocking :meth:`Balance.get_filter_mode`."""
    return self._portal.call(self._bal.get_filter_mode)

def get_isocal_mode(self) -> IsoCalMode:
    """Blocking :meth:`Balance.get_isocal_mode`."""
    return self._portal.call(self._bal.get_isocal_mode)

def get_menu_access(self) -> MenuAccessMode:
    """Blocking :meth:`Balance.get_menu_access`."""
    return self._portal.call(self._bal.get_menu_access)

def get_tare_behavior(self) -> TareBehavior:
    """Blocking :meth:`Balance.get_tare_behavior`."""
    return self._portal.call(self._bal.get_tare_behavior)

def identify(self) -> DeviceInfo:
    """Blocking :meth:`Balance.identify`."""
    return self._portal.call(self._bal.identify)

def increment(self, area: int = 0) -> Quantity:
    """Blocking :meth:`Balance.increment`."""
    return self._portal.call(self._bal.increment, area)

def internal_adjust(
    self,
    *,
    cal_type: int | None = None,
    confirm: bool = False,
) -> None:
    """Blocking :meth:`Balance.internal_adjust`."""
    self._portal.call(
        self._bal.internal_adjust,
        cal_type=cal_type,
        confirm=confirm,
    )

def last_cal_record(self) -> CalRecord:
    """Blocking :meth:`Balance.last_cal_record`."""
    return self._portal.call(self._bal.last_cal_record)

def poll(self) -> Reading:
    """Blocking :meth:`Balance.poll`."""
    return self._portal.call(self._bal.poll)

def raw_sbi(
    self,
    command: bytes | str,
    *,
    confirm: bool = False,
    timeout: float | None = None,
    expect_lines: int = 1,
) -> SbiReply:
    """Blocking :meth:`Balance.raw_sbi`."""
    return self._portal.call(
        self._bal.raw_sbi,
        command,
        confirm=confirm,
        timeout=timeout,
        expect_lines=expect_lines,
    )

def raw_xbpi(
    self,
    opcode: int,
    args: bytes = b"",
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> XbpiFrame:
    """Blocking :meth:`Balance.raw_xbpi`."""
    return self._portal.call(
        self._bal.raw_xbpi,
        opcode,
        args,
        confirm=confirm,
        timeout=timeout,
    )

def read_gross(self, *, hires: int = 0) -> Reading:
    """Blocking :meth:`Balance.read_gross`."""
    return self._portal.call(self._bal.read_gross, hires=hires)

def read_net(self, *, hires: int = 0) -> Reading:
    """Blocking :meth:`Balance.read_net`."""
    return self._portal.call(self._bal.read_net, hires=hires)

def read_parameter(self, index: int) -> ParameterEntry:
    """Blocking :meth:`Balance.read_parameter`."""
    return self._portal.call(self._bal.read_parameter, index)

def read_tare_value(self) -> Reading:
    """Blocking :meth:`Balance.read_tare_value`."""
    return self._portal.call(self._bal.read_tare_value)

def refresh_sbi_autoprint_state(self, *, timeout: float | None = None) -> bool:
    """Blocking :meth:`Balance.refresh_sbi_autoprint_state`."""
    return self._portal.call(
        self._bal.refresh_sbi_autoprint_state,
        timeout=timeout,
    )

def reload_menu(self, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.reload_menu`."""
    self._portal.call(self._bal.reload_menu, confirm=confirm)

def save_menu(self, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.save_menu`."""
    self._portal.call(self._bal.save_menu, confirm=confirm)

def set_auto_zero(self, mode: AutoZeroMode | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_auto_zero`."""
    self._portal.call(self._bal.set_auto_zero, mode, confirm=confirm)

def set_baud_rate(
    self,
    wire_code: int,
    *,
    baudrate: int,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
    timeout: float | None = None,
    confirm: bool = False,
) -> DeviceInfo:
    """Blocking :meth:`Balance.set_baud_rate`."""
    return self._portal.call(
        self._bal.set_baud_rate,
        wire_code,
        baudrate=baudrate,
        parity=parity,
        stopbits=stopbits,
        timeout=timeout,
        confirm=confirm,
    )

def set_display_unit(self, unit: Unit | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_display_unit`."""
    self._portal.call(self._bal.set_display_unit, unit, confirm=confirm)

def set_filter_mode(self, mode: FilterMode | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_filter_mode`."""
    self._portal.call(self._bal.set_filter_mode, mode, confirm=confirm)

def set_isocal_mode(self, mode: IsoCalMode | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_isocal_mode`."""
    self._portal.call(self._bal.set_isocal_mode, mode, confirm=confirm)

def set_menu_access(self, mode: MenuAccessMode | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_menu_access`."""
    self._portal.call(self._bal.set_menu_access, mode, confirm=confirm)

def set_tare_behavior(self, mode: TareBehavior | str | int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.set_tare_behavior`."""
    self._portal.call(self._bal.set_tare_behavior, mode, confirm=confirm)

def status(self) -> BalanceStatus:
    """Blocking :meth:`Balance.status`."""
    return self._portal.call(self._bal.status)

def tare(self) -> None:
    """Blocking :meth:`Balance.tare`."""
    self._portal.call(self._bal.tare)

def temperature(self, sensor: int = 0) -> TemperatureReading:
    """Blocking :meth:`Balance.temperature`."""
    return self._portal.call(self._bal.temperature, sensor)

def write_parameter(self, index: int, value: int, *, confirm: bool = False) -> None:
    """Blocking :meth:`Balance.write_parameter`."""
    self._portal.call(self._bal.write_parameter, index, value, confirm=confirm)

def write_sbn_address(
    self,
    sbn: int,
    *,
    update_session_dst: bool = False,
    timeout: float | None = None,
    confirm: bool = False,
) -> int:
    """Blocking :meth:`Balance.write_sbn_address`."""
    return self._portal.call(
        self._bal.write_sbn_address,
        sbn,
        update_session_dst=update_session_dst,
        timeout=timeout,
        confirm=confirm,
    )

def zero(self) -> None:
    """Blocking :meth:`Balance.zero`."""
    self._portal.call(self._bal.zero)

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.
    """
    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."""
    self._require_portal()
    return SyncAsyncIterator(self, async_iter)

def add(
    self,
    name: str,
    source: SyncBalance | Balance | str | Transport,
    *,
    protocol: ProtocolKind | None = None,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
    src_sbn: int = 0x01,
    dst_sbn: int = 0x09,
    strict: bool = False,
    identify: bool = True,
) -> SyncBalance:
    """Blocking :meth:`SartoriusManager.add`.

    Accepts a :class:`SyncBalance` as ``source`` in addition to
    the async shapes — the wrapper is unwrapped to the underlying
    :class:`Balance` before delegation.
    """
    from sartoriuslib.protocol.base import ProtocolKind as _ProtocolKind  # noqa: PLC0415

    effective_protocol = protocol if protocol is not None else _ProtocolKind.XBPI
    mgr = self._require_mgr()
    async_source: Balance | str | Transport = unwrap_sync_balance(source)
    async_balance = self.portal.call(
        mgr.add,
        name,
        async_source,
        protocol=effective_protocol,
        serial_settings=serial_settings,
        timeout=timeout,
        src_sbn=src_sbn,
        dst_sbn=dst_sbn,
        strict=strict,
        identify=identify,
    )
    wrapped = wrap_balance(async_balance, self.portal)
    self._wrapped[name] = wrapped
    return wrapped

def close(self) -> None:
    """Blocking :meth:`SartoriusManager.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:`SartoriusManager.execute`."""
    mgr = self._require_mgr()
    return self.portal.call(mgr.execute, command, requests_by_name)

def get(self, name: str) -> SyncBalance:
    """Return the sync wrapper for the balance registered under ``name``."""
    cached = self._wrapped.get(name)
    if cached is not None:
        return cached
    mgr = self._require_mgr()
    async_balance = mgr.get(name)
    wrapped = wrap_balance(async_balance, self.portal)
    self._wrapped[name] = wrapped
    return wrapped

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

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

def close(self) -> None:
    """Blocking :meth:`SampleSink.close` — idempotent."""
    portal = self._portal
    if portal is None:
        return
    portal.call(self._async_sink.close)

def open(self) -> None:
    """Blocking :meth:`SampleSink.open`."""
    self.portal.call(self._async_sink.open)

def write_many(self, samples: Sequence[Sample]) -> None:
    """Blocking :meth:`SampleSink.write_many`."""
    self.portal.call(self._async_sink.write_many, samples)