runtimepy.channel package

Subpackages

• runtimepy.channel.environment package
  • Subpackages
    • runtimepy.channel.environment.command package
      • Submodules
      • runtimepy.channel.environment.command.parser module
      • runtimepy.channel.environment.command.processor module
      • runtimepy.channel.environment.command.result module
      • Module contents
  • Submodules
  • runtimepy.channel.environment.array module
    • ArrayChannelEnvironment
      • ArrayChannelEnvironment.array()
      • ArrayChannelEnvironment.names
    • ChannelArray
      • ChannelArray.array
      • ChannelArray.create()
      • ChannelArray.names
  • runtimepy.channel.environment.base module
    • BaseChannelEnvironment
      • BaseChannelEnvironment.add_field()
      • BaseChannelEnvironment.add_int()
      • BaseChannelEnvironment.age_ns()
      • BaseChannelEnvironment.apply()
      • BaseChannelEnvironment.exists()
      • BaseChannelEnvironment.field_or_channel()
      • BaseChannelEnvironment.finalize()
      • BaseChannelEnvironment.get()
      • BaseChannelEnvironment.get_bool()
      • BaseChannelEnvironment.get_float()
      • BaseChannelEnvironment.get_int()
      • BaseChannelEnvironment.sample_bool_for()
      • BaseChannelEnvironment.sample_enum_for()
      • BaseChannelEnvironment.sample_float_for()
      • BaseChannelEnvironment.sample_int_for()
      • BaseChannelEnvironment.set()
      • BaseChannelEnvironment.value()
      • BaseChannelEnvironment.values()
      • BaseChannelEnvironment.wait_for_bool()
      • BaseChannelEnvironment.wait_for_enum()
      • BaseChannelEnvironment.wait_for_numeric()
      • BaseChannelEnvironment.wait_for_numeric_isclose()
  • runtimepy.channel.environment.create module
    • CreateChannelEnvironment
      • CreateChannelEnvironment.bool_channel()
      • CreateChannelEnvironment.channel()
      • CreateChannelEnvironment.enum()
      • CreateChannelEnvironment.float_channel()
      • CreateChannelEnvironment.int_channel()
  • runtimepy.channel.environment.file module
    • FileChannelEnvironment
      • FileChannelEnvironment.export()
      • FileChannelEnvironment.export_directory()
      • FileChannelEnvironment.export_json()
      • FileChannelEnvironment.load()
      • FileChannelEnvironment.load_directory()
      • FileChannelEnvironment.load_json()
  • runtimepy.channel.environment.sample module
    • long_name_int_enum()
    • poll_sample_env()
    • random()
    • sample_bool_enum()
    • sample_env()
    • sample_fields()
    • sample_float()
    • sample_int_enum()
  • runtimepy.channel.environment.telemetry module
    • TelemetryChannelEnvironment
      • TelemetryChannelEnvironment.events
      • TelemetryChannelEnvironment.ingest()
      • TelemetryChannelEnvironment.parse_event_stream()
      • TelemetryChannelEnvironment.registered()
  • Module contents
    • ChannelEnvironment
      • ChannelEnvironment.num_defaults
      • ChannelEnvironment.search_names()
      • ChannelEnvironment.set_default()
• runtimepy.channel.event package
  • Submodules
  • runtimepy.channel.event.header module
    • PrimitiveEventHeader
      • PrimitiveEventHeader.init_header()
      • PrimitiveEventHeader.initialize()
  • Module contents
    • PrimitiveEvent
      • PrimitiveEvent.registered()
      • PrimitiveEvent.set_min_period()

Submodules

runtimepy.channel.registry module

A module implementing a channel registry.

class runtimepy.channel.registry.ChannelNameRegistry(mapping: MutableMapping[int, str] = None, reverse: MutableMapping[str, int] = None)

Bases: NameRegistry

A name registry with a name-matching pattern for channel names.

name_regex: _Pattern | None = re.compile('^[a-zA-Z0-9_.-]+$')

class runtimepy.channel.registry.ChannelRegistry(data: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]] = None, schemas: SchemaMap = None, dest_attr: str = 'data', verify: bool = True)

Bases: Registry[Channel[Any]]

A runtime enumeration registry.

channel(name: str, kind: Primitive[Any] | type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str, commandable: bool = False, enum: str | int = None, scaling: list[float | int] = None, description: str = None, default: int | float | bool | None = None, controls: dict[str, int | float | bool | dict[str, int | float | bool]] | str = None, **kwargs) → Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive] | None

Create a new channel.

event_fifo: ByteFifo

event_header: Protocol

header_ready: bool

init(data: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]) → None

Perform implementation-specific initialization.

property kind: type[Channel[Any]]

Determine what kind of registry this is.

name_registry

alias of ChannelNameRegistry

registered(stream: BinaryIO, pattern: str = '.*', exact: bool = False, flush: bool = False, channel: ChannelMetrics = None) → Iterator[list[str]]

Register a stream as a managed context. Returns a list of all channels registered.

class runtimepy.channel.registry.ParsedEvent(name: str, timestamp: int, value: bool | int | float)

Bases: NamedTuple

A raw channel event.

static by_channel(event_stream: Iterable[ParsedEvent]) → dict[str, list[ParsedEvent]]

Get a dictionary of channel events broken down by individual channels.

name: str

Alias for field number 0

timestamp: int

Alias for field number 1

value: bool | int | float

Alias for field number 2

Module contents

A module implementing a basic channel interface.

class runtimepy.channel.Channel(data: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]] = None, schemas: SchemaMap = None, dest_attr: str = 'data', verify: bool = True)

Bases: RegistryItem, EnumMixin, Generic[T]

An interface for an individual channel.

asdict() → dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]

Obtain a dictionary representing this instance.

default: int | float | bool | None

property has_default: bool

Determine if this channel has a default value configured.

init(data: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]) → None

Perform implementation-specific initialization.

raw: T

set_default(default: int | float | bool | None = None) → None

Set a new default value for this channel.

property type: Int8Type | Int16Type | Int32Type | Int64Type | Uint8Type | Uint16Type | Uint32Type | Uint64Type | HalfType | FloatType | DoubleType | BooleanType

Get the underlying primitive type of this channel.

update_primitive(primitive: T) → None

Use a new underlying primitive for this channel.