channel

Channel information.

Channel dataclass

Channel(name, data_type, sample_rate, time=None, publisher=None, partition_id=None, partition_index=None, stride=None, max_latency=None, replay_id=None)

Metadata associated with a channel.

Channels have the form {domain}:*.

Parameters:

Name	Type	Description	Default
name	str	The name associated with this channel.	required
data_type	str	The data type associated with this channel.	required
sample_rate	float	The sampling rate associated with this channel.	required
time	int	The GPS timestamp when this metadata became active, in nanoseconds.	None
publisher	str	The publisher associated with this channel.	None
partition_id	str	The Kafka partition ID associated with this channel.	None
partition_index	int	Partition index for the channel. It is unique within the partition and allows the use of an integer value to identify the channel instead of a string.	None
stride	int	Time duration in individual blocks of data for this channel, in nanoseconds.	None
max_latency	int	Maximum expected publication latency for this channel's data, in nanoseconds.	None

as_dict

as_dict()

Return metadata as "serialized" dict

Source code in arrakis/channel.py

def as_dict(self) -> dict[str, Any]:
    """Return metadata as "serialized" dict"""
    return asdict(self)

fields staticmethod

fields()

Channel field names

Source code in arrakis/channel.py

@staticmethod
def fields() -> tuple[str, ...]:
    """Channel field names"""
    return tuple(field.name for field in fields(Channel))

from_field classmethod

from_field(field)

Create a Channel from Arrow Flight field metadata.

Parameters:

Name	Type	Description	Default
field	Field	The channel field containing relevant metadata.	required

Returns:

Type	Description
Channel	The newly created channel.

Source code in arrakis/channel.py

@classmethod
def from_field(cls, field: pyarrow.Field) -> Channel:
    """Create a Channel from Arrow Flight field metadata.

    Parameters
    ----------
    field : pyarrow.Field
        The channel field containing relevant metadata.

    Returns
    -------
    Channel
        The newly created channel.

    """
    data_type = _list_dtype_to_str(field.type)
    assert field.metadata is not None
    sample_rate = float(field.metadata[b"rate"].decode())
    return cls(field.name, data_type, sample_rate)

from_json classmethod

from_json(payload)

Create a Channel from its JSON representation.

Parameters:

Name	Type	Description	Default
payload	str	The JSON-serialized channel.	required

Returns:

Type	Description
Channel	The newly created channel.

Source code in arrakis/channel.py

@classmethod
def from_json(cls, payload: str) -> Channel:
    """Create a Channel from its JSON representation.

    Parameters
    ----------
    payload : str
        The JSON-serialized channel.

    Returns
    -------
    Channel
        The newly created channel.

    """
    obj = json.loads(payload)
    return cls(**obj)

from_validated classmethod

from_validated(name, data_type, sample_rate, time=None, publisher=None, partition_id=None, partition_index=None, stride=None, max_latency=None, replay_id=None)

Create a Channel from already-validated field values.

Skips the name validation and dtype normalization performed by the regular constructor, which dominate construction cost when loading large channel sets in bulk. Callers must guarantee the values came from a validated source (e.g. a metadata cache populated through the regular constructor): name must be a valid channel name and data_type a numpy dtype name string. parsed_name is still computed (and validated) on first access.

Source code in arrakis/channel.py

@classmethod
def from_validated(
    cls,
    name: str,
    data_type: str,
    sample_rate: float,
    time: int | None = None,
    publisher: str | None = None,
    partition_id: str | None = None,
    partition_index: int | None = None,
    stride: int | None = None,
    max_latency: int | None = None,
    replay_id: str | None = None,
) -> Channel:
    """Create a Channel from already-validated field values.

    Skips the name validation and dtype normalization performed by
    the regular constructor, which dominate construction cost when
    loading large channel sets in bulk.  Callers must guarantee
    the values came from a validated source (e.g. a metadata cache
    populated through the regular constructor): ``name`` must be a
    valid channel name and ``data_type`` a numpy dtype name
    string.  ``parsed_name`` is still computed (and validated) on
    first access.
    """
    channel = object.__new__(cls)
    channel.__dict__.update(
        name=name,
        data_type=data_type,
        sample_rate=sample_rate,
        time=time,
        publisher=publisher,
        partition_id=partition_id,
        partition_index=partition_index,
        stride=stride,
        max_latency=max_latency,
        replay_id=replay_id,
    )
    return channel

ParsedChannelName dataclass

ParsedChannelName(domain, subsystem, subsystem_delimiter, rest)

A parsed channel name

Channel names have the following structure:

<domain>:<subsystem>[-_]<rest>

parse classmethod

parse(name)

Parse a channel name into it's constituent parts

Source code in arrakis/channel.py

@classmethod
def parse(cls, name: str) -> ParsedChannelName:
    """Parse a channel name into it's constituent parts"""
    if rem := CHANNEL_NAME_RE.match(name):
        return cls(
            rem["domain"],
            rem["subsystem"],
            rem["delimiter"],
            rem["rest"],
        )
    msg = f"Invalid channel name format: '{name}'"
    raise ValueError(msg)