runtimepy.channel.environment package

Subpackages

• runtimepy.channel.environment.command package
  • Submodules
  • runtimepy.channel.environment.command.parser module
    • ChannelCommand
      • ChannelCommand.GET
      • ChannelCommand.SET
      • ChannelCommand.TOGGLE
    • CommandParser
      • CommandParser.data
      • CommandParser.error()
      • CommandParser.exit()
      • CommandParser.initialize()
  • runtimepy.channel.environment.command.processor module
    • ChannelCommandProcessor
      • ChannelCommandProcessor.command()
      • ChannelCommandProcessor.do_set()
      • ChannelCommandProcessor.do_toggle()
      • ChannelCommandProcessor.get_suggestion()
      • ChannelCommandProcessor.handle_command()
      • ChannelCommandProcessor.parse()
    • RemoteCommandProcessor
      • RemoteCommandProcessor.handle_command()
  • runtimepy.channel.environment.command.result module
    • CommandResult
      • CommandResult.reason
      • CommandResult.success
  • Module contents
    • GlobalEnvironment
      • GlobalEnvironment.clear()
      • GlobalEnvironment.event_telemetry_output()
      • GlobalEnvironment.export()
      • GlobalEnvironment.file_event_stream()
      • GlobalEnvironment.from_root()
      • GlobalEnvironment.log_file()
      • GlobalEnvironment.meta_path()
      • GlobalEnvironment.read_event_stream()
      • GlobalEnvironment.register()
      • GlobalEnvironment.temporary()
      • GlobalEnvironment.valid_root
      • GlobalEnvironment.write_meta()
    • clear_env()
    • register_env()

Submodules

runtimepy.channel.environment.array module

A channel-environment extension for creating arrays of primitives.

class runtimepy.channel.environment.array.ArrayChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: BaseChannelEnvironment

A channel-environment extension for working with arrays of primitives.

array(keys: Iterable[str | int]) → ChannelArray

Create a primitive array from an in-order iterable of registry keys.

class runtimepy.channel.environment.array.ChannelArray(names: list[str], array: PrimitiveArray)

Bases: NamedTuple

A class for managing an array of channels and bit-fields.

array: PrimitiveArray

Alias for field number 1

static create() → ChannelArray

Create a new, empty channel array.

names: list[str]

Alias for field number 0

runtimepy.channel.environment.base module

A module implementing a base channel environment.

class runtimepy.channel.environment.base.BaseChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: NamespaceMixin, FinalizeMixin

A class integrating channel and enumeration registries.

add_field(field: BitField, namespace: Namespace = None) → str

Add a bit field to the environment.

add_int(key: str | int, amount: int) → int

Modify an integer channel.

age_ns(key: str | int) → int

Get the age of an entity based on registry key.

apply(values: dict[str | int, bool | int | float | str]) → None

Apply a map of values to the environment.

exists(val: str | int) → bool

Determine if a channel exists.

field_or_channel(val: str | int) → BitField | Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive] | None

Attempt to look up a field or channel for a given registry key.

finalize(strict: bool = True) → None

Finalize this instance.

get(val: str | int) → tuple[Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive], RuntimeEnum | None] | None

Attempt to get a channel and its enumeration (if it has one).

get_bool(key: str | int) → tuple[Channel[BooleanPrimitive], RuntimeEnum | None]

Get a boolean channel.

get_float(key: str | int) → Channel[FloatPrimitive] | Channel[DoublePrimitive]

Get a floating-point channel.

get_int(key: str | int) → tuple[Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive], RuntimeEnum | None]

Get an integer channel.

set(key: str | int, value: bool | int | float | str) → None

Attempt to set an arbitrary channel value.

value(key: str | int, resolve_enum: bool = True) → bool | int | float | str

Attempt to get a channel’s current value.

values(resolve_enum: bool = True) → dict[str | int, bool | int | float | str]

Get a new dictionary of current channel values.

runtimepy.channel.environment.create module

A module for creating channels at runtime.

class runtimepy.channel.environment.create.CreateChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: BaseChannelEnvironment

An environment extension for creating channels.

bool_channel(name: str, kind: Primitive[Any] | type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str = 'bool', commandable: bool = False, enum: str | int | RuntimeEnum = None, namespace: Namespace = None, **kwargs) → tuple[Channel[BooleanPrimitive], RuntimeEnum | None]

Create a boolean channel.

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 | RuntimeEnum = None, namespace: Namespace = None, scaling: list[float | int] = None, min_period_s: float = None, **kwargs) → tuple[Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive] | Channel[FloatPrimitive] | Channel[DoublePrimitive] | Channel[BooleanPrimitive], RuntimeEnum | None]

Create a new channel from the environment.

enum(name: str, kind: str | EnumType, items: MutableMapping[str | int, str | int] | MutableMapping[str | bool, str | bool] | MutableMapping[str, bool] | MutableMapping[str, int] = None, namespace: Namespace = None, primitive: str = 'uint8') → RuntimeEnum

Create a new enum from the environment.

float_channel(name: str, kind: Primitive[Any] | type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str = 'float', commandable: bool = False, namespace: Namespace = None, scaling: list[float | int] = None, **kwargs) → Channel[FloatPrimitive] | Channel[DoublePrimitive]

Create a floating-point channel.

int_channel(name: str, kind: Primitive[Any] | type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str = 'uint32', commandable: bool = False, enum: str | int | RuntimeEnum = None, namespace: Namespace = None, scaling: list[float | int] = None, **kwargs) → tuple[Channel[Int8Primitive] | Channel[Int16Primitive] | Channel[Int32Primitive] | Channel[Int64Primitive] | Channel[Uint8Primitive] | Channel[Uint16Primitive] | Channel[Uint32Primitive] | Channel[Uint64Primitive], RuntimeEnum | None]

Create an integer channel.

runtimepy.channel.environment.file module

A channel-environment extension for loading and saving files.

class runtimepy.channel.environment.file.FileChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: BaseChannelEnvironment

A class integrating file-system operations with channel environments.

export(channels: Path | str | None = 'channels.json', enums: Path | str | None = 'enums.json', values: Path | str | None = 'values.json', fields: Path | str | None = 'fields.json', names: Path | str | None = 'names.json', resolve_enum: bool = True, **kwargs) → None

Write channel and enum registries to disk.

export_directory(path: Path | str | None, resolve_enum: bool = True, **kwargs) → None

Export this channel environment to a directory.

export_json(resolve_enum: bool = True) → dict[str, Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]]]

Get this channel environment as a single dictionary.

classmethod load(channels: Path | str | None = 'channels.json', enums: Path | str | None = 'enums.json', values: Path | str | None = 'values.json', fields: Path | str | None = 'fields.json', names: Path | str | None = 'names.json', finalize: bool = True) → T

Load a channel environment from a pair of files.

classmethod load_directory(path: Path | str | None, finalize: bool = True) → T

Load a channel environment from a directory.

classmethod load_json(data: dict[str, Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]]], finalize: bool = True) → T

Load a channel environment from JSON data.

runtimepy.channel.environment.sample module

Interfaces for working with example environment configurations.

runtimepy.channel.environment.sample.poll_sample_env(env: ChannelEnvironment) → None

Register sample enumerations and channels to an environment.

runtimepy.channel.environment.sample.random() → x in the interval [0, 1).

runtimepy.channel.environment.sample.sample_env(env: ChannelEnvironment = None) → ChannelEnvironment

Register sample enumerations and channels to an environment.

runtimepy.channel.environment.telemetry module

A module implementing channel-environment telemetry registration.

class runtimepy.channel.environment.telemetry.TelemetryChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: BaseChannelEnvironment

A class integrating telemetry streaming.

events: list[str]

ingest(point: ParsedEvent) → None

Update internal state based on an event. Note that the event timestamp is not respected.

parse_event_stream(stream: BinaryIO) → Iterator[ParsedEvent]

Parse individual events from a stream.

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.

Module contents

A module implementing a channel environment.

class runtimepy.channel.environment.ChannelEnvironment(channels: ChannelRegistry = None, enums: EnumRegistry = None, values: dict[str | int, bool | int | float | str] = None, fields: Iterable[BitFields] = None, namespace: Namespace = None, namespace_delim: str = '.')

Bases: TelemetryChannelEnvironment, ArrayChannelEnvironment, FileChannelEnvironment, CreateChannelEnvironment

A class integrating channel and enumeration registries.

property names: Iterator[str]

Iterate over registered names in the environment.