| | |
- runtimepy.channel.environment.base.BaseChannelEnvironment(vcorelib.namespace.mixin.NamespaceMixin, runtimepy.mixins.finalize.FinalizeMixin)
-
- TelemetryChannelEnvironment
class TelemetryChannelEnvironment(runtimepy.channel.environment.base.BaseChannelEnvironment) |
| |
TelemetryChannelEnvironment(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
A class integrating telemetry streaming.
|
| |
- Method resolution order:
- TelemetryChannelEnvironment
- runtimepy.channel.environment.base.BaseChannelEnvironment
- vcorelib.namespace.mixin.NamespaceMixin
- runtimepy.mixins.finalize.FinalizeMixin
- builtins.object
Methods defined here:
- 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.
Data and other attributes defined here:
- __annotations__ = {'events': list[str]}
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.
| |