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)

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)