| | |
- builtins.tuple(builtins.object)
-
- ParsedEvent
- runtimepy.registry.Registry(runtimepy.schemas.RuntimepyDictCodec, typing.Generic)
-
- ChannelRegistry
- runtimepy.registry.name.NameRegistry(runtimepy.mapping.TwoWayNameMapping)
-
- ChannelNameRegistry
class ChannelNameRegistry(runtimepy.registry.name.NameRegistry) |
| |
ChannelNameRegistry(mapping: MutableMapping[int, str] = None, reverse: MutableMapping[str, int] = None) -> None
A name registry with a name-matching pattern for channel names.
|
| |
- Method resolution order:
- ChannelNameRegistry
- runtimepy.registry.name.NameRegistry
- runtimepy.mapping.TwoWayNameMapping
- runtimepy.mixins.regex.RegexMixin
- vcorelib.logging.LoggerMixin
- typing.Generic
- builtins.object
Data and other attributes defined here:
- __annotations__ = {}
- __parameters__ = ()
- name_regex = re.compile('^[a-z0-9_.-]+$')
Methods inherited from runtimepy.registry.name.NameRegistry:
- __init__(self, mapping: MutableMapping[int, str] = None, reverse: MutableMapping[str, int] = None) -> None
- Initialize this name registry.
- register_name(self, name: str, identifier: int = None) -> Optional[int]
- Register a new name.
Data and other attributes inherited from runtimepy.registry.name.NameRegistry:
- __orig_bases__ = (runtimepy.mapping.TwoWayNameMapping[int],)
Methods inherited from runtimepy.mapping.TwoWayNameMapping:
- asdict(self) -> dict[str, ~T]
- Provide a dictionary representation.
- identifier(self, key: Union[str, ~T]) -> Optional[~T]
- Get the integer identifier associated with a registry key.
- load_key_to_name(self, mapping: MutableMapping[~T, str]) -> None
- Load a key-to-name mapping.
- load_name_to_key(self, reverse: MutableMapping[str, ~T]) -> None
- Load a name-to-key mapping.
- name(self, key: Union[str, ~T]) -> Optional[str]
- Get the name associated with a registry key.
- search(self, pattern: str, exact: bool = False) -> Iterator[str]
- Get names in this mapping based on a pattern.
Class methods inherited from runtimepy.mapping.TwoWayNameMapping:
- bool_from_dict(data: MutableMapping[Union[str, bool], Union[str, bool]]) -> ~BoolMapping from builtins.type
- Create a boolean-to-name mapping from a dictionary with arbitrary data.
- int_from_dict(data: MutableMapping[Union[str, int], Union[str, int]]) -> ~IntMapping from builtins.type
- Create an integer-to-name mapping from a dictionary with arbitrary
data.
Readonly properties inherited from runtimepy.mapping.TwoWayNameMapping:
- names
- Iterate over names.
Class methods inherited from runtimepy.mixins.regex.RegexMixin:
- validate_name(name: str, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None) -> bool from builtins.type
- Determine if a name is valid for this class.
Data descriptors inherited from runtimepy.mixins.regex.RegexMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Class methods inherited from typing.Generic:
- __class_getitem__(...) from builtins.type
- Parameterizes a generic class.
At least, parameterizing a generic class is the *main* thing this
method does. For example, for some generic class `Foo`, this is called
when we do `Foo[int]` - there, with `cls=Foo` and `params=int`.
However, note that this method is also called when defining generic
classes in the first place with `class Foo[T]: ...`.
- __init_subclass__(...) from builtins.type
- Function to initialize subclasses.
|
class ChannelRegistry(runtimepy.registry.Registry) |
| |
ChannelRegistry(data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, schemas: vcorelib.schemas.base.SchemaMap = None, dest_attr: str = 'data', verify: bool = True) -> None
A runtime enumeration registry.
|
| |
- Method resolution order:
- ChannelRegistry
- runtimepy.registry.Registry
- runtimepy.schemas.RuntimepyDictCodec
- vcorelib.dict.codec.DictCodec
- vcorelib.dict.codec.JsonCodec
- abc.ABC
- vcorelib.schemas.mixins.SchemaMixin
- typing.Generic
- builtins.object
Methods defined here:
- 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] = None, scaling: list[typing.Union[float, int]] = None, description: str = None) -> 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], NoneType]
- Create a new channel.
- init(self, data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]]) -> None
- Perform implementation-specific initialization.
- 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.
Readonly properties defined here:
- kind
- Determine what kind of registry this is.
Data and other attributes defined here:
- __abstractmethods__ = frozenset()
- __annotations__ = {'event_fifo': <class 'vcorelib.io.fifo.ByteFifo'>, 'event_header': <class 'runtimepy.codec.protocol.Protocol'>, 'header_ready': <class 'bool'>}
- __orig_bases__ = (runtimepy.registry.Registry[runtimepy.channel.Channel[typing.Any]],)
- __parameters__ = ()
- name_registry = <class 'runtimepy.channel.registry.ChannelNameRegistry'>
- A name registry with a name-matching pattern for channel names.
Methods inherited from runtimepy.registry.Registry:
- __getitem__(self, key: Union[str, int]) -> ~T
- Get a registry item.
- asdict(self) -> Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]]
- Get this registry as a dictionary.
- get(self, key: Union[str, int]) -> Optional[~T]
- Attempt to get an item from a registry key.
- register(self, name: str, item: ~T) -> bool
- Attempt to register a new item.
- register_dict(self, name: str, data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]]) -> Optional[~T]
- Register a new item from dictionary data.
- search(self, pattern: str, exact: bool = False) -> Iterator[tuple[str, ~T]]
- Search for items in the registry by name.
Data and other attributes inherited from runtimepy.schemas.RuntimepyDictCodec:
- default_schemas = {'StructConfig': <vcorelib.schemas.json.JsonSche...chemas.json.JsonSchema object at 0x7f08e9b518e0>}
Methods inherited from vcorelib.dict.codec.DictCodec:
- __init__(self, data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, schemas: vcorelib.schemas.base.SchemaMap = None, dest_attr: str = 'data', verify: bool = True) -> None
- Initialize this instance.
Class methods inherited from vcorelib.dict.codec.DictCodec:
- create(data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, schemas: vcorelib.schemas.base.SchemaMap = None, dest_attr: str = 'data', verify: bool = True) -> ~V from abc.ABCMeta
- A factory method to create an instance.
Methods inherited from vcorelib.dict.codec.JsonCodec:
- __copy__(self: ~T) -> ~T
- Create another instance based on this instance's data.
- __eq__(self, other) -> bool
- Determine if this instance is equal to another.
- __str__(self) -> str
- Use the dictionary representation of this instance for string
representation.
- encode(self, pathlike: Union[pathlib.Path, str, NoneType], arbiter: vcorelib.io.arbiter.DataArbiter = <vcorelib.io.arbiter.DataArbiter object at 0x7f08ea687590>, **kwargs) -> Tuple[bool, int]
- Encode this object instance to a file.
Class methods inherited from vcorelib.dict.codec.JsonCodec:
- decode(pathlike: Union[pathlib.Path, str, NoneType], arbiter: vcorelib.io.arbiter.DataArbiter = <vcorelib.io.arbiter.DataArbiter object at 0x7f08ea687590>, schemas: vcorelib.schemas.base.SchemaMap = None, dest_attr: str = 'data', verify: bool = True, require_success: bool = True, default_data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, **kwargs) -> ~T from abc.ABCMeta
- Decode an object instance from data loaded from a file.
- file_cache(pathlike: Union[pathlib.Path, str, NoneType], arbiter: vcorelib.io.arbiter.DataArbiter = <vcorelib.io.arbiter.DataArbiter object at 0x7f08ea687590>, encode_kwargs: Dict[str, Any] = None, require_success: bool = False, **kwargs) -> Iterator[~T] from abc.ABCMeta
- Manage an instance of this class as a disk-backed file.
- normalize(data: Union[Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], ~T]) -> ~T from abc.ABCMeta
- Ensure that some object is an instance of this class.
Data descriptors inherited from vcorelib.dict.codec.JsonCodec:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
Data and other attributes inherited from vcorelib.dict.codec.JsonCodec:
- __hash__ = None
Methods inherited from vcorelib.schemas.mixins.SchemaMixin:
- validate(self, valid_attr: str = 'data') -> None
- Validate an instance attribute based on this instance's schema.
Class methods inherited from vcorelib.schemas.mixins.SchemaMixin:
- schema_name() -> str from abc.ABCMeta
- A default name for this class's schema.
Class methods inherited from typing.Generic:
- __class_getitem__(...) from abc.ABCMeta
- Parameterizes a generic class.
At least, parameterizing a generic class is the *main* thing this
method does. For example, for some generic class `Foo`, this is called
when we do `Foo[int]` - there, with `cls=Foo` and `params=int`.
However, note that this method is also called when defining generic
classes in the first place with `class Foo[T]: ...`.
- __init_subclass__(...) from abc.ABCMeta
- Function to initialize subclasses.
|
class ParsedEvent(builtins.tuple) |
| |
ParsedEvent(name: str, timestamp: int, value: Union[bool, int, float])
A raw channel event.
|
| |
- Method resolution order:
- ParsedEvent
- builtins.tuple
- builtins.object
Methods defined here:
- __getnewargs__(self)
- Return self as a plain tuple. Used by copy and pickle.
- __repr__(self)
- Return a nicely formatted representation string
- _asdict(self)
- Return a new dict which maps field names to their values.
- _replace(self, /, **kwds)
- Return a new ParsedEvent object replacing specified fields with new values
Class methods defined here:
- _make(iterable) from builtins.type
- Make a new ParsedEvent object from a sequence or iterable
Static methods defined here:
- __new__(_cls, name: str, timestamp: int, value: Union[bool, int, float])
- Create new instance of ParsedEvent(name, timestamp, value)
- by_channel(event_stream: Iterable[ForwardRef('ParsedEvent')]) -> dict[str, list['ParsedEvent']]
- Get a dictionary of channel events broken down by individual channels.
Data descriptors defined here:
- name
- Alias for field number 0
- timestamp
- Alias for field number 1
- value
- Alias for field number 2
Data and other attributes defined here:
- __annotations__ = {'name': <class 'str'>, 'timestamp': <class 'int'>, 'value': typing.Union[bool, int, float]}
- __match_args__ = ('name', 'timestamp', 'value')
- __orig_bases__ = (<function NamedTuple>,)
- _field_defaults = {}
- _fields = ('name', 'timestamp', 'value')
Methods inherited from builtins.tuple:
- __add__(self, value, /)
- Return self+value.
- __contains__(self, key, /)
- Return bool(key in self).
- __eq__(self, value, /)
- Return self==value.
- __ge__(self, value, /)
- Return self>=value.
- __getattribute__(self, name, /)
- Return getattr(self, name).
- __getitem__(self, key, /)
- Return self[key].
- __gt__(self, value, /)
- Return self>value.
- __hash__(self, /)
- Return hash(self).
- __iter__(self, /)
- Implement iter(self).
- __le__(self, value, /)
- Return self<=value.
- __len__(self, /)
- Return len(self).
- __lt__(self, value, /)
- Return self<value.
- __mul__(self, value, /)
- Return self*value.
- __ne__(self, value, /)
- Return self!=value.
- __rmul__(self, value, /)
- Return value*self.
- count(self, value, /)
- Return number of occurrences of value.
- index(self, value, start=0, stop=9223372036854775807, /)
- Return first index of value.
Raises ValueError if the value is not present.
Class methods inherited from builtins.tuple:
- __class_getitem__(...) from builtins.type
- See PEP 585
| |