fujilib.streaming¶

@classmethod
def from_error(
    cls,
    error: FujiError,
    *,
    device: str,
    address: int,
    protocol: ProtocolKind,
    timing: TransferTiming,
    metadata: Mapping[str, str] | None = None,
    channels: Iterable[ChannelId] = (),
) -> Sample:
    """Build the sample of a failed poll, timed by the failed attempt.

    Pass the recording's ``channels`` so the sample's row has the same
    columns as a successful one.
    """
    return cls(
        device=device,
        address=address,
        frame=None,
        protocol=protocol,
        t_mono_ns=timing.midpoint_mono_ns,
        t_utc=timing.midpoint_utc,
        requested_at=timing.requested_at,
        received_at=timing.received_at,
        latency_s=timing.latency_s,
        metadata=MappingProxyType(dict(metadata or {})),
        error=error,
        channels=tuple(channels),
    )

@classmethod
def from_frame(
    cls,
    frame: Frame,
    *,
    device: str,
    address: int,
    metadata: Mapping[str, str] | None = None,
    channels: Iterable[ChannelId] | None = None,
) -> Sample:
    """Build the sample of a successful poll, timed by its concentration block.

    ``channels`` defaults to the frame's own.
    """
    timing = frame.readings_timing
    return cls(
        device=device,
        address=address,
        frame=frame,
        protocol=frame.protocol,
        t_mono_ns=timing.midpoint_mono_ns,
        t_utc=timing.midpoint_utc,
        requested_at=timing.requested_at,
        received_at=timing.received_at,
        latency_s=timing.latency_s,
        metadata=MappingProxyType(dict(metadata or {})),
        channels=frame.channels if channels is None else tuple(channels),
    )

@classmethod
def failure(cls, error: FujiError) -> Self:
    """A failed result holding ``error``."""
    return cls(value=None, error=error)

@classmethod
def success[V](cls, value: V) -> DeviceResult[V]:
    """A successful result holding ``value``."""
    del cls
    return DeviceResult(value=value, error=None)

def layout(self, names: Sequence[str] | None = None) -> Mapping[str, SourceLayout]:
    """Describe the named analyzers (all when ``None``), with no I/O."""
    ...

async def poll(self, names: Sequence[str] | None = None) -> Mapping[str, DeviceResult[Frame]]:
    """Poll the named analyzers (all when ``None``) once; a failure is a failed result."""
    ...

async def reconnect(self, name: str) -> None:
    """Open analyzer ``name`` again after a connection failure.

    Raises:
        FujiError: it could not be opened, or is not the analyzer it was.
    """
    ...

def layout(self, names: Sequence[str] | None = None) -> Mapping[str, SourceLayout]:
    """The analyzer's station, protocol and established channels, under its name.

    An empty mapping when ``names`` leaves this analyzer out.
    """
    if not self._selected(names):
        return MappingProxyType({})
    device = self._device
    layout = SourceLayout(
        address=device.address,
        protocol=device.protocol,
        channels=tuple(c.channel for c in device.channels),
        reopenable=device.session.reopenable,
    )
    return MappingProxyType({self._name: layout})

async def poll(self, names: Sequence[str] | None = None) -> Mapping[str, DeviceResult[Frame]]:
    """Poll the analyzer once: two transactions.

    Returns ``{name: result}``, or an empty mapping when ``names`` leaves
    this analyzer out. A failed poll is a failed result, not an exception.
    """
    if not self._selected(names):
        return MappingProxyType({})
    result: DeviceResult[Frame]
    try:
        result = DeviceResult.success(await self._device.poll())
    except FujiError as exc:
        result = DeviceResult(value=None, error=exc)
    return MappingProxyType({self._name: result})

async def reconnect(self, name: str) -> None:
    """Reopen the analyzer (:meth:`Analyzer.reopen <fujilib.devices.analyzer.Analyzer.reopen>`).

    Raises:
        FujiValidationError: ``name`` is not this source's name.
        FujiError: the analyzer could not be reopened.
    """
    if name != self._name:
        msg = f"this source publishes {self._name!r}, not {name!r}"
        raise FujiValidationError(msg)
    _ = await self._device.reopen()

def __post_init__(self) -> None:
    """Check the schedule.

    Raises:
        FujiValidationError: an empty or negative schedule, or fewer than one attempt.
    """
    if not self.backoff_s or not all(_is_seconds(s) and s >= 0 for s in self.backoff_s):
        msg = f"backoff_s must be one or more finite, non-negative seconds: {self.backoff_s!r}"
        raise FujiValidationError(msg)
    attempts = self.max_attempts
    if attempts is not None and (not _is_int(attempts) or attempts < 1):
        msg = f"max_attempts must be None or at least 1, got {attempts!r}"
        raise FujiValidationError(msg)

def delay(self, attempt: int) -> float:
    """The wait before attempt ``attempt`` (1 for the first) of an outage."""
    return self.backoff_s[min(attempt, len(self.backoff_s)) - 1]

def __aiter__(self) -> AsyncIterator[T]:
    """Iterate the batches of :attr:`stream`."""
    return self.stream.__aiter__()