watlowlib.sync¶

def close(self) -> None:
    """Cancel the underlying async iterator if it exposes ``aclose``."""
    if self._closed:
        return
    self._closed = True
    if not self._portal.running:
        return
    aclose: Callable[[], Awaitable[Any]] | None = getattr(self._aiter, "aclose", None)
    if aclose is None:
        return
    with contextlib.suppress(Exception):
        self._portal.call(aclose)

def close(self) -> None:
    """Blocking :meth:`Controller.aclose`. Idempotent."""
    if not self._portal.running:
        return
    self._portal.call(self._ctl.aclose)

def identify(self, *, timeout: float | None = None) -> DeviceInfo:
    """Blocking :meth:`Controller.identify`."""
    return self._portal.call(self._ctl.identify, timeout=timeout)

def loop(self, n: int) -> SyncControllerLoop:
    """Return a sync sub-facade bound to loop ``n`` (1-indexed)."""
    return SyncControllerLoop(self._ctl.loop(n), self._portal)

def poll(
    self,
    parameters: Sequence[str | int],
    *,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
) -> list[Sample]:
    """Blocking :meth:`Controller.poll`."""
    return self._portal.call(
        self._ctl.poll,
        parameters,
        names=names,
        instances=instances,
    )

def read_parameter(
    self,
    name_or_id: str | int,
    *,
    instance: int = 1,
    timeout: float | None = None,
) -> ParameterEntry:
    """Blocking :meth:`Controller.read_parameter`."""
    return self._portal.call(
        self._ctl.read_parameter,
        name_or_id,
        instance=instance,
        timeout=timeout,
    )

def read_pv(self, *, instance: int = 1, timeout: float | None = None) -> Reading:
    """Blocking :meth:`Controller.read_pv`."""
    return self._portal.call(self._ctl.read_pv, instance=instance, timeout=timeout)

def read_setpoint(self, *, instance: int = 1, timeout: float | None = None) -> Reading:
    """Blocking :meth:`Controller.read_setpoint`."""
    return self._portal.call(self._ctl.read_setpoint, instance=instance, timeout=timeout)

def set_setpoint(
    self,
    value: float,
    *,
    instance: int = 1,
    confirm: bool = False,
    timeout: float | None = None,
) -> Reading:
    """Blocking :meth:`Controller.set_setpoint`."""
    return self._portal.call(
        self._ctl.set_setpoint,
        value,
        instance=instance,
        confirm=confirm,
        timeout=timeout,
    )

def write_parameter(
    self,
    name_or_id: str | int,
    value: float | int | str,
    *,
    instance: int = 1,
    confirm: bool = False,
    timeout: float | None = None,
) -> ParameterEntry:
    """Blocking :meth:`Controller.write_parameter`."""
    return self._portal.call(
        self._ctl.write_parameter,
        name_or_id,
        value,
        instance=instance,
        confirm=confirm,
        timeout=timeout,
    )

def read_alarms(self) -> AlarmState:
    """Blocking :meth:`ControllerLoop.read_alarms`."""
    return self._portal.call(self._loop.read_alarms)

def read_output(self) -> Reading:
    """Blocking :meth:`ControllerLoop.read_output`."""
    return self._portal.call(self._loop.read_output)

def read_pid(self) -> PidGains:
    """Blocking :meth:`ControllerLoop.read_pid`."""
    return self._portal.call(self._loop.read_pid)

def read_pv(self, *, timeout: float | None = None) -> Reading:
    """Blocking :meth:`ControllerLoop.read_pv`."""
    return self._portal.call(self._loop.read_pv, timeout=timeout)

def read_setpoint(self, *, timeout: float | None = None) -> Reading:
    """Blocking :meth:`ControllerLoop.read_setpoint`."""
    return self._portal.call(self._loop.read_setpoint, timeout=timeout)

def set_setpoint(
    self,
    value: float,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> Reading:
    """Blocking :meth:`ControllerLoop.set_setpoint`."""
    return self._portal.call(
        self._loop.set_setpoint,
        value,
        confirm=confirm,
        timeout=timeout,
    )

def write_pid(self, gains: PidGains, *, confirm: bool = False) -> PidGains:
    """Blocking :meth:`ControllerLoop.write_pid`."""
    return self._portal.call(self._loop.write_pid, gains, confirm=confirm)

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 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)

def add(
    self,
    name: str,
    source: SyncController | Controller | str | Transport,
    *,
    protocol: ProtocolKind = ProtocolKind.STDBUS,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    family: ControllerFamily = ControllerFamily.UNKNOWN,
) -> SyncController:
    """Blocking :meth:`WatlowManager.add`.

    Accepts a :class:`SyncController` as ``source`` in addition to
    the async shapes — the wrapper is unwrapped to the underlying
    :class:`Controller` before delegation.
    """
    mgr = self._require_mgr()
    async_source: Controller | str | Transport = unwrap_sync_controller(source)
    async_controller = self.portal.call(
        mgr.add,
        name,
        async_source,
        protocol=protocol,
        address=address,
        serial_settings=serial_settings,
        family=family,
    )
    wrapped = wrap_controller(async_controller, self.portal)
    self._wrapped[name] = wrapped
    return wrapped

def close(self) -> None:
    """Blocking :meth:`WatlowManager.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_each[T](
    self,
    op: Callable[[Controller], Awaitable[T]],
    names: Sequence[str] | None = None,
) -> dict[str, DeviceResult[T]]:
    """Blocking :meth:`WatlowManager.execute_each`.

    ``op`` receives the **async** :class:`Controller` so existing
    coroutines compose. If you have a sync helper, wrap it in an
    async stub or run it on the portal yourself.
    """
    mgr = self._require_mgr()
    return self.portal.call(mgr.execute_each, op, names)

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

def poll(
    self,
    parameters: Sequence[str | int],
    *,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
) -> list[Sample]:
    """Blocking :meth:`WatlowManager.poll`."""
    mgr = self._require_mgr()
    return self.portal.call(
        mgr.poll,
        parameters,
        names=names,
        instances=instances,
    )

def record_to_sink(
    self,
    *,
    parameters: Sequence[str | int],
    rate_hz: float,
    duration: float | None = None,
    sink: SyncSinkAdapter | SampleSink,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
    overflow: OverflowPolicy | None = None,
    buffer_size: int = 64,
    batch_size: int = 64,
    flush_interval: float = 1.0,
) -> AcquisitionSummary:
    """Record polled samples directly into a sink — one-call convenience.

    Combines :func:`watlowlib.sync.record` and
    :func:`watlowlib.sync.pipe` into a single blocking call. The
    manager's portal is reused for both legs so the recorder and
    the sink share an event loop. ``sink`` may be either a
    :class:`SyncSinkAdapter` (preferred, opened externally) or a
    bare async :class:`SampleSink` — in the latter case this
    method opens the sink against the manager's portal and closes
    it after the recording finishes.

    Returns the :class:`AcquisitionSummary` from
    :func:`watlowlib.sync.pipe`.
    """
    # Lazy imports — sink machinery pulls heavy deps (anyio sink
    # primitives, sqlite, etc.) and we want the surface importable
    # without that until the user reaches for streaming.
    from watlowlib.streaming.recorder import OverflowPolicy as _OverflowPolicy  # noqa: PLC0415
    from watlowlib.sync.recording import pipe, record  # noqa: PLC0415
    from watlowlib.sync.sinks import SyncSinkAdapter  # noqa: PLC0415

    self._require_mgr()
    active_overflow = overflow if overflow is not None else _OverflowPolicy.BLOCK
    portal = self.portal

    with ExitStack() as stack:
        sink_for_pipe: SyncSinkAdapter | SampleSink
        if isinstance(sink, SyncSinkAdapter):
            # Caller-owned sync wrapper — no open / close here.
            sink_for_pipe = sink
        else:
            # Bare async sink — wrap on this manager's portal so it
            # shares the recorder's event loop, and own the
            # open/close lifecycle through the ExitStack.
            wrapped = SyncSinkAdapter(sink, portal=portal)
            stack.enter_context(wrapped)
            sink_for_pipe = wrapped

        stream = stack.enter_context(
            record(
                self,
                parameters=parameters,
                rate_hz=rate_hz,
                duration=duration,
                names=names,
                instances=instances,
                overflow=active_overflow,
                buffer_size=buffer_size,
                portal=portal,
            ),
        )
        return pipe(
            stream,
            sink_for_pipe,
            batch_size=batch_size,
            flush_interval=flush_interval,
            portal=portal,
        )

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

@staticmethod
@contextmanager
def open(
    port: str,
    *,
    protocol: ProtocolKind | None = None,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    portal: SyncPortal | None = None,
) -> Generator[SyncController]:
    """Open a sync :class:`SyncController` scoped to a ``with`` block.

    Mirrors :func:`watlowlib.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=``.
    """
    effective_protocol = protocol if protocol is not None else ProtocolKind.STDBUS

    with ExitStack() as stack:
        active_portal = portal if portal is not None else stack.enter_context(SyncPortal())
        controller = active_portal.call(
            open_device,
            port,
            protocol=effective_protocol,
            address=address,
            serial_settings=serial_settings,
        )
        # ``open_device`` returns a controller that may or may not
        # already be open: AUTO returned by the detector is open;
        # STDBUS / MODBUS_RTU need ``__aenter__`` to run open().
        # ``Controller.__aenter__`` short-circuits when already open
        # and returns ``self``; calling it through the portal here
        # gives us the same lifecycle as ``async with`` does.
        active_portal.call(controller.__aenter__)
        try:
            yield wrap_controller(controller, active_portal)
        finally:
            # Close the underlying transport through the portal.
            active_portal.call(controller.aclose)

def close(self) -> None:
    """Cancel the underlying async iterator if it exposes ``aclose``."""
    if self._closed:
        return
    self._closed = True
    if not self._portal.running:
        return
    aclose: Callable[[], Awaitable[Any]] | None = getattr(self._aiter, "aclose", None)
    if aclose is None:
        return
    with contextlib.suppress(Exception):
        self._portal.call(aclose)

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 close(self) -> None:
    """Blocking :meth:`Controller.aclose`. Idempotent."""
    if not self._portal.running:
        return
    self._portal.call(self._ctl.aclose)

def identify(self, *, timeout: float | None = None) -> DeviceInfo:
    """Blocking :meth:`Controller.identify`."""
    return self._portal.call(self._ctl.identify, timeout=timeout)