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)

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 timestamp when this metadata became active.	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 seconds.	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)
    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)