| |
- Method resolution order:
- ChannelEnvironment
- runtimepy.channel.environment.telemetry.TelemetryChannelEnvironment
- runtimepy.channel.environment.array.ArrayChannelEnvironment
- runtimepy.channel.environment.file.FileChannelEnvironment
- runtimepy.channel.environment.create.CreateChannelEnvironment
- runtimepy.channel.environment.base.BaseChannelEnvironment
- vcorelib.namespace.mixin.NamespaceMixin
- runtimepy.mixins.finalize.FinalizeMixin
- builtins.object
Readonly properties defined here:
- names
- Iterate over registered names in the environment.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from runtimepy.channel.environment.telemetry.TelemetryChannelEnvironment:
- ingest(self, point: runtimepy.channel.registry.ParsedEvent) -> None
- Update internal state based on an event. Note that the event timestamp
is not respected.
- parse_event_stream(self, stream: <class 'BinaryIO'>) -> Iterator[runtimepy.channel.registry.ParsedEvent]
- Parse individual events from a stream.
- registered(self, stream: <class 'BinaryIO'>, pattern: str = '.*', exact: bool = False, flush: bool = False, channel: runtimepy.metrics.channel.ChannelMetrics = None) -> Iterator[list[str]]
- Register a stream as a managed context. Returns a list of all channels
registered.
Methods inherited from runtimepy.channel.environment.array.ArrayChannelEnvironment:
- array(self, keys: Iterable[Union[str, int]]) -> runtimepy.channel.environment.array.ChannelArray
- Create a primitive array from an in-order iterable of registry keys.
Methods inherited from runtimepy.channel.environment.file.FileChannelEnvironment:
- export(self, channels: Union[pathlib.Path, str, NoneType] = 'channels.json', enums: Union[pathlib.Path, str, NoneType] = 'enums.json', values: Union[pathlib.Path, str, NoneType] = 'values.json', fields: Union[pathlib.Path, str, NoneType] = 'fields.json', names: Union[pathlib.Path, str, NoneType] = 'names.json', resolve_enum: bool = True, **kwargs) -> None
- Write channel and enum registries to disk.
- export_directory(self, path: Union[pathlib.Path, str, NoneType], resolve_enum: bool = True, **kwargs) -> None
- Export this channel environment to a directory.
- export_json(self, resolve_enum: bool = True) -> dict[str, typing.Dict[str, typing.Union[str, int, float, bool, NoneType, typing.Dict[str, typing.Union[str, int, float, bool, NoneType]], typing.List[typing.Union[str, int, float, bool, NoneType]]]]]
- Get this channel environment as a single dictionary.
Class methods inherited from runtimepy.channel.environment.file.FileChannelEnvironment:
- load(channels: Union[pathlib.Path, str, NoneType] = 'channels.json', enums: Union[pathlib.Path, str, NoneType] = 'enums.json', values: Union[pathlib.Path, str, NoneType] = 'values.json', fields: Union[pathlib.Path, str, NoneType] = 'fields.json', names: Union[pathlib.Path, str, NoneType] = 'names.json', finalize: bool = True) -> ~T from builtins.type
- Load a channel environment from a pair of files.
- load_directory(path: Union[pathlib.Path, str, NoneType], finalize: bool = True) -> ~T from builtins.type
- Load a channel environment from a directory.
- load_json(data: dict[str, typing.Dict[str, typing.Union[str, int, float, bool, NoneType, typing.Dict[str, typing.Union[str, int, float, bool, NoneType]], typing.List[typing.Union[str, int, float, bool, NoneType]]]]], finalize: bool = True) -> ~T from builtins.type
- Load a channel environment from JSON data.
Methods inherited from runtimepy.channel.environment.create.CreateChannelEnvironment:
- bool_channel(self, name: str, kind: Union[runtimepy.primitives.base.Primitive[Any], type[Union[runtimepy.primitives.int.Int8Primitive, runtimepy.primitives.int.Int16Primitive, runtimepy.primitives.int.Int32Primitive, runtimepy.primitives.int.Int64Primitive, runtimepy.primitives.int.Uint8Primitive, runtimepy.primitives.int.Uint16Primitive, runtimepy.primitives.int.Uint32Primitive, runtimepy.primitives.int.Uint64Primitive, runtimepy.primitives.float.HalfPrimitive, runtimepy.primitives.float.FloatPrimitive, runtimepy.primitives.float.DoublePrimitive, runtimepy.primitives.bool.BooleanPrimitive]], str] = 'bool', commandable: bool = False, enum: Union[str, int, runtimepy.enum.RuntimeEnum] = None, namespace: vcorelib.namespace.base.Namespace = None, **kwargs) -> tuple[runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Create a boolean channel.
- channel(self, name: str, kind: Union[runtimepy.primitives.base.Primitive[Any], type[Union[runtimepy.primitives.int.Int8Primitive, runtimepy.primitives.int.Int16Primitive, runtimepy.primitives.int.Int32Primitive, runtimepy.primitives.int.Int64Primitive, runtimepy.primitives.int.Uint8Primitive, runtimepy.primitives.int.Uint16Primitive, runtimepy.primitives.int.Uint32Primitive, runtimepy.primitives.int.Uint64Primitive, runtimepy.primitives.float.HalfPrimitive, runtimepy.primitives.float.FloatPrimitive, runtimepy.primitives.float.DoublePrimitive, runtimepy.primitives.bool.BooleanPrimitive]], str], commandable: bool = False, enum: Union[str, int, runtimepy.enum.RuntimeEnum] = None, namespace: vcorelib.namespace.base.Namespace = None, scaling: list[typing.Union[float, int]] = None, min_period_s: float = None, **kwargs) -> tuple[typing.Union[runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive], runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive], runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive]], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Create a new channel from the environment.
- enum(self, name: str, kind: Union[str, ForwardRef('EnumType')], items: Union[MutableMapping[Union[str, int], Union[str, int]], MutableMapping[Union[str, bool], Union[str, bool]], MutableMapping[str, bool], MutableMapping[str, int]] = None, namespace: vcorelib.namespace.base.Namespace = None, primitive: str = 'uint8') -> runtimepy.enum.RuntimeEnum
- Create a new enum from the environment.
- float_channel(self, name: str, kind: Union[runtimepy.primitives.base.Primitive[Any], type[Union[runtimepy.primitives.int.Int8Primitive, runtimepy.primitives.int.Int16Primitive, runtimepy.primitives.int.Int32Primitive, runtimepy.primitives.int.Int64Primitive, runtimepy.primitives.int.Uint8Primitive, runtimepy.primitives.int.Uint16Primitive, runtimepy.primitives.int.Uint32Primitive, runtimepy.primitives.int.Uint64Primitive, runtimepy.primitives.float.HalfPrimitive, runtimepy.primitives.float.FloatPrimitive, runtimepy.primitives.float.DoublePrimitive, runtimepy.primitives.bool.BooleanPrimitive]], str] = 'float', commandable: bool = False, namespace: vcorelib.namespace.base.Namespace = None, scaling: list[typing.Union[float, int]] = None, **kwargs) -> Union[runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive]]
- Create a floating-point channel.
- int_channel(self, name: str, kind: Union[runtimepy.primitives.base.Primitive[Any], type[Union[runtimepy.primitives.int.Int8Primitive, runtimepy.primitives.int.Int16Primitive, runtimepy.primitives.int.Int32Primitive, runtimepy.primitives.int.Int64Primitive, runtimepy.primitives.int.Uint8Primitive, runtimepy.primitives.int.Uint16Primitive, runtimepy.primitives.int.Uint32Primitive, runtimepy.primitives.int.Uint64Primitive, runtimepy.primitives.float.HalfPrimitive, runtimepy.primitives.float.FloatPrimitive, runtimepy.primitives.float.DoublePrimitive, runtimepy.primitives.bool.BooleanPrimitive]], str] = 'uint32', commandable: bool = False, enum: Union[str, int, runtimepy.enum.RuntimeEnum] = None, namespace: vcorelib.namespace.base.Namespace = None, scaling: list[typing.Union[float, int]] = None, **kwargs) -> tuple[typing.Union[runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive]], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Create an integer channel.
Methods inherited from runtimepy.channel.environment.base.BaseChannelEnvironment:
- __eq__(self, other) -> bool
- Determine if two channel environments are equivalent.
- __getitem__(self, key: Union[str, int]) -> tuple[typing.Union[runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive], runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive], runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive]], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Get a channel and its enumeration.
- __init__(self, channels: runtimepy.channel.registry.ChannelRegistry = None, enums: runtimepy.enum.registry.EnumRegistry = None, values: dict[typing.Union[str, int], typing.Union[bool, int, float, str]] = None, fields: Iterable[runtimepy.primitives.field.fields.BitFields] = None, namespace: vcorelib.namespace.base.Namespace = None, namespace_delim: str = '.') -> None
- Initialize this channel environment.
- __setitem__(self, key: Union[str, int], value: Union[bool, int, float, str]) -> None
- Mapping-set interface.
- add_field(self, field: runtimepy.primitives.field.BitField, namespace: vcorelib.namespace.base.Namespace = None) -> str
- Add a bit field to the environment.
- add_int(self, key: Union[str, int], amount: int) -> int
- Modify an integer channel.
- age_ns(self, key: Union[str, int]) -> int
- Get the age of an entity based on registry key.
- apply(self, values: dict[typing.Union[str, int], typing.Union[bool, int, float, str]]) -> None
- Apply a map of values to the environment.
- exists(self, val: Union[str, int]) -> bool
- Determine if a channel exists.
- field_or_channel(self, val: Union[str, int]) -> Union[runtimepy.primitives.field.BitField, runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive], runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive], runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive], NoneType]
- Attempt to look up a field or channel for a given registry key.
- finalize(self, strict: bool = True) -> None
- Finalize this instance.
- get(self, val: Union[str, int]) -> Optional[tuple[Union[runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive], runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive], runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive]], Optional[runtimepy.enum.RuntimeEnum]]]
- Attempt to get a channel and its enumeration (if it has one).
- get_bool(self, key: Union[str, int]) -> tuple[runtimepy.channel.Channel[runtimepy.primitives.bool.BooleanPrimitive], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Get a boolean channel.
- get_float(self, key: Union[str, int]) -> Union[runtimepy.channel.Channel[runtimepy.primitives.float.FloatPrimitive], runtimepy.channel.Channel[runtimepy.primitives.float.DoublePrimitive]]
- Get a floating-point channel.
- get_int(self, key: Union[str, int]) -> tuple[typing.Union[runtimepy.channel.Channel[runtimepy.primitives.int.Int8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Int64Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint8Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint16Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint32Primitive], runtimepy.channel.Channel[runtimepy.primitives.int.Uint64Primitive]], typing.Optional[runtimepy.enum.RuntimeEnum]]
- Get an integer channel.
- set(self, key: Union[str, int], value: Union[bool, int, float, str]) -> None
- Attempt to set an arbitrary channel value.
- value(self, key: Union[str, int], resolve_enum: bool = True) -> Union[bool, int, float, str]
- Attempt to get a channel's current value.
- values(self, resolve_enum: bool = True) -> dict[typing.Union[str, int], typing.Union[bool, int, float, str]]
- Get a new dictionary of current channel values.
Data and other attributes inherited from runtimepy.channel.environment.base.BaseChannelEnvironment:
- __hash__ = None
Methods inherited from vcorelib.namespace.mixin.NamespaceMixin:
- child_namespace(self, *names: str, namespace: vcorelib.namespace.base.Namespace = None) -> vcorelib.namespace.base.Namespace
- Obtain a child namespace.
- names_pushed(self, *names: str, namespace: vcorelib.namespace.base.Namespace = None) -> Iterator[NoneType]
- Apply some names to this object's namespace as a managed context.
- namespace(self, name: str = None, delim: str = None, namespace: vcorelib.namespace.base.Namespace = None) -> str
- Get a namespace string for this object.
- namespace_search(self, *names: str, pattern: str = '.*', namespace: vcorelib.namespace.base.Namespace = None) -> Iterator[str]
- Perform a search on the namespace.
- namespace_suggest(self, data: str, delta: bool = True, namespace: vcorelib.namespace.base.Namespace = None) -> Optional[str]
- Find the shortest name suggestion.
- pop_name(self, name: str = None, namespace: vcorelib.namespace.base.Namespace = None) -> str
- Pop the latest name off the stack.
- push_name(self, name: str, namespace: vcorelib.namespace.base.Namespace = None) -> None
- Push a name onto the stack.
Readonly properties inherited from vcorelib.namespace.mixin.NamespaceMixin:
- ns
- Get this instance's namespace.
Data descriptors inherited from vcorelib.namespace.mixin.NamespaceMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
Methods inherited from runtimepy.mixins.finalize.FinalizeMixin:
- async wait_finalized(self) -> None
- Wait for this instance to be finalized.
Readonly properties inherited from runtimepy.mixins.finalize.FinalizeMixin:
- finalized
- Determine if this instance is finalized or not.
|