channel

Channel information.

Channel dataclass

Channel(name, data_type, sample_rate, time=None, publisher=None, partition_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	dtype	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 partition ID associated with this channel.	None

domain cached property

domain

The domain associated with this 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 = numpy.dtype(_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	bytes	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: bytes) -> Channel:
    """Create a Channel from its JSON representation.

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

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

    """
    obj = json.loads(payload)
    obj["data_type"] = numpy.dtype(obj["data_type"])
    return cls(**obj)

to_json

to_json(time=None)

Serialize channel metadata to JSON.

Parameters:

Name	Type	Description	Default
time	int	If specified, the timestamp when this metadata became active.	None

Source code in arrakis/channel.py

def to_json(self, time: int | None = None) -> str:
    """Serialize channel metadata to JSON.

    Parameters
    ----------
    time : int, optional
        If specified, the timestamp when this metadata became active.

    """
    # generate dict from dataclass and adjust fields
    # to be JSON compatible. In addition, store the
    # channel name, as well as updating the timestamp
    # if passed in.
    obj = asdict(self)
    obj["data_type"] = numpy.dtype(self.data_type).name
    if time is not None:
        obj["time"] = time
    return json.dumps(obj)