runtimepy.primitives package

runtimepy.primitives package#

Subpackages#

Submodules#

runtimepy.primitives.base module#

A module implementing a base, primitive-type storage entity.

class runtimepy.primitives.base.Primitive(value: ~runtimepy.primitives.base.T = None, scaling: list[float | int] = None, time_source: ~typing.Callable[[], int] = <function default_time_ns>)[source]#

Bases: Generic[T]

A simple class for storing an underlying primitive value.

age_ns(now: int = None) → int[source]#: Get the age of this primitive’s value in nanoseconds.

age_str(now: int = None) → str[source]#: Get the age of this primitive’s value as a string.

binary(byte_order: ByteOrder = None) → bytes[source]#: Convert this instance to a byte array.

byte_order: ByteOrder = 4#

callback(callback: Callable[[T, T], None]) → Iterator[None][source]#: Register a callback as a managed context.

copy() → Primitive[T][source]#: A simple wrapper for copy.

classmethod decode(data: bytes | bytearray | memoryview, byte_order: ByteOrder = None) → T[source]#: Decode a primitive of this type from provided data.

classmethod encode(value: T, byte_order: ByteOrder = None) → bytes[source]#: Create a bytes instance based on this primitive type.

from_stream(stream: BinaryIO, byte_order: ByteOrder = None) → T[source]#: Update this primitive from a stream and return the new value.

invert(value: T) → float | int[source]#: Invert a value using this primitive’s scaling.

kind: Int8Type | Int16Type | Int32Type | Int64Type | Uint8Type | Uint16Type | Uint32Type | Uint64Type | HalfType | FloatType | DoubleType | BooleanType#

classmethod read(stream: BinaryIO, byte_order: ByteOrder = None) → T[source]#: Read a primitive from the provided stream based on this primitive type.

register_callback(callback: Callable[[T, T], None], once: bool = False) → int[source]#: Register a callback and return an identifier for it.

remove_callback(callback_id: int) → bool[source]#: Remove a callback if one is registered with this identifier.

scale(value: T) → float | int[source]#: Scale a value using this primitive’s scaling.

property scaled: float | int#: Get this primitive as a scaled value.

set_value(value: T, timestamp_ns: int = None) → None[source]#: Set a new underlying value.

property size: int#: Get the size of this primitive.

to_stream(stream: BinaryIO, byte_order: ByteOrder = None) → int[source]#: Write this primitive to a stream.

update(data: bytes | bytearray | memoryview, byte_order: ByteOrder = None) → T[source]#: Update this primitive from a bytes object.

classmethod valid_primitive(primitive: bool | int | float) → bool[source]#: Determine if a Python primitive is valid for this class.

property value: T#: Obtain the underlying value.

classmethod write(value: T, stream: BinaryIO, byte_order: ByteOrder = None) → int[source]#: Write a primitive to the stream based on this type.

runtimepy.primitives.bool module#

A module implementing a boolean-primitive interface.

runtimepy.primitives.bool.Bool#: alias of BooleanPrimitive

class runtimepy.primitives.bool.BooleanPrimitive(value: bool = False, **kwargs)[source]#

Bases: Primitive[bool]

A simple primitive class for booleans.

clear(timestamp_ns: int = None) → None[source]#: Coerce the underlying value to false.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.bool.BooleanType object>#

randomize(timestamp_ns: int = None) → None[source]#: Set this primitive to a random integer.

set(timestamp_ns: int = None) → None[source]#: Coerce the underlying value to true.

toggle(timestamp_ns: int = None) → None[source]#: Toggle the underlying value.

async wait_for_state(state: bool, timeout: float) → EvalResult[source]#: Wait for this primitive to reach a specified state.

runtimepy.primitives.bool.getrandbits(k) → x. Generates an int with k random bits.#

runtimepy.primitives.byte_order module#

A module implementing an enumeration for byte ordering options.

class runtimepy.primitives.byte_order.ByteOrder(*values)[source]#

Bases: RuntimeIntEnum

An enumeration for viable byte orders.

BIG_ENDIAN = 3#

LITTLE_ENDIAN = 2#

NATIVE = 1#

NETWORK = 4#

property fmt: str#: Get the struct formatter for this byte order.

classmethod id() → int | None[source]#: Override in sub-class to coerce enum id.

runtimepy.primitives.byte_order.enum_registry(*kinds: type[RuntimeIntEnum], register_byte_order: bool = True) → EnumRegistry[source]#: Create an enum registry with the provided custom types registered.

runtimepy.primitives.evaluation module#

A module implementing interfaces for evaluating the underlying state of primitives.

class runtimepy.primitives.evaluation.EvalResult(*values)[source]#

Bases: RuntimeIntEnum

A container for all possible evaluation results.

FAIL = 3#

NOT_SET = 1#

SUCCESS = 4#

TIMEOUT = 2#

class runtimepy.primitives.evaluation.Operator(*values)[source]#

Bases: RuntimeIntEnum

https://docs.python.org/3/library/operator.html.

EQUAL = 3#

GREATER_THAN = 6#

GREATER_THAN_OR_EQUAL = 5#

LESS_THAN = 1#

LESS_THAN_OR_EQUAL = 2#

NOT_EQUAL = 4#

class runtimepy.primitives.evaluation.PrimitiveIsCloseMixin(value: ~runtimepy.primitives.base.T = None, scaling: list[float | int] = None, time_source: ~typing.Callable[[], int] = <function default_time_ns>)[source]#

Bases: Primitive[T]

Adds a wait-for-isclose method.

async wait_for_isclose(value: float, timeout: float, rel_tol: float = 1e-09, abs_tol: float = 0.0) → EvalResult[source]#: Wait for this primitive to reach a specified state.

async runtimepy.primitives.evaluation.compare_latest(primitive: Primitive[T], lhs: T, timeout: float, operation: Operator = Operator.EQUAL) → EvalResult[source]#: Perform a canonical comparison of the latest value with a provided value.

async runtimepy.primitives.evaluation.evaluate(primitive: Primitive[T], evaluator: Callable[[T, T], EvalResult], timeout: float) → EvalResult[source]#: Evaluate a primitive within a timeout constraint and return the result.

async runtimepy.primitives.evaluation.sample_for(primitive: Primitive[T], timeout: float, count: int = -1, current: bool = True) → AsyncIterator[tuple[T, int]][source]#: Sample a primitive until timeout or ‘count’ samples are emitted (if ‘count’ is set).

runtimepy.primitives.float module#

A module implementing a floating-point primitive interface.

class runtimepy.primitives.float.BaseFloatPrimitive(value: float = 0.0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: PrimitiveIsCloseMixin[float]

A simple primitive class for floating-point numbers.

randomize(timestamp_ns: int = None) → None[source]#: Set this primitive to a random integer.

async wait_for_value(value: float, timeout: float, operation: Operator = Operator.EQUAL) → EvalResult[source]#: Wait for this primitive to reach a specified state.

runtimepy.primitives.float.Double#: alias of DoublePrimitive

class runtimepy.primitives.float.DoublePrimitive(value: float = 0.0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseFloatPrimitive

A simple primitive class for double-precision floating-point.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.float.DoubleType object>#

runtimepy.primitives.float.Float#: alias of FloatPrimitive

class runtimepy.primitives.float.FloatPrimitive(value: float = 0.0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseFloatPrimitive

A simple primitive class for single-precision floating-point.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.float.FloatType object>#

runtimepy.primitives.float.Half#: alias of HalfPrimitive

class runtimepy.primitives.float.HalfPrimitive(value: float = 0.0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseFloatPrimitive

A simple primitive class for half-precision floating-point.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.float.HalfType object>#

runtimepy.primitives.float.random() → x in the interval [0, 1).#

runtimepy.primitives.int module#

A module implementing an integer-primitive interface.

class runtimepy.primitives.int.BaseIntPrimitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: PrimitiveIsCloseMixin[int]

A simple primitive class for integer primitives.

increment(amount: int = 1, timestamp_ns: int = None) → int[source]#: Increment this primitive by some amount and return the new value.

randomize(timestamp_ns: int = None) → None[source]#: Set this primitive to a random integer.

async wait_for_increment(timeout: float, count: int = 1) → EvalResult[source]#: Wait for this primitive to increment by a certain amount.

async wait_for_value(value: int | float, timeout: float, operation: Operator = Operator.EQUAL) → EvalResult[source]#: Wait for this primitive to reach a specified state.

runtimepy.primitives.int.Int16#: alias of Int16Primitive

class runtimepy.primitives.int.Int16Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

A signed 16-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Int16Type object>#

runtimepy.primitives.int.Int32#: alias of Int32Primitive

class runtimepy.primitives.int.Int32Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

A signed 32-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Int32Type object>#

runtimepy.primitives.int.Int64#: alias of Int64Primitive

class runtimepy.primitives.int.Int64Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

A signed 64-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Int64Type object>#

runtimepy.primitives.int.Int8#: alias of Int8Primitive

class runtimepy.primitives.int.Int8Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

A signed 8-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Int8Type object>#

runtimepy.primitives.int.Uint16#: alias of Uint16Primitive

class runtimepy.primitives.int.Uint16Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

An unsigned 16-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Uint16Type object>#

runtimepy.primitives.int.Uint32#: alias of Uint32Primitive

class runtimepy.primitives.int.Uint32Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

An unsigned 32-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Uint32Type object>#

runtimepy.primitives.int.Uint64#: alias of Uint64Primitive

class runtimepy.primitives.int.Uint64Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

An unsigned 64-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Uint64Type object>#

runtimepy.primitives.int.Uint8#: alias of Uint8Primitive

class runtimepy.primitives.int.Uint8Primitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: BaseIntPrimitive

An unsigned 8-bit primitive.

kind: _AnyPrimitiveType = <runtimepy.primitives.types.int.Uint8Type object>#

runtimepy.primitives.scaling module#

A module implementing interfaces for working with channel scaling polynomials.

runtimepy.primitives.scaling.apply(value: float | int, scaling: list[float | int] = None) → float | int[source]#: Apply a scaling polynomial to a value.

runtimepy.primitives.scaling.invert(value: float | int, scaling: list[float | int] = None, should_round: bool = False) → float | int[source]#: Apply a scaling polynomial to a value.

runtimepy.primitives.string module#

A module implementing integer-prefixed string reading and writing.

class runtimepy.primitives.string.StringPrimitive(value: str = '', kind: type[~runtimepy.primitives.base.Primitive[int]] = <class 'runtimepy.primitives.int.Uint16Primitive'>, byte_order: ~runtimepy.primitives.byte_order.ByteOrder = ByteOrder.NETWORK)[source]#

Bases: object

A class implementing a string-primitive interface.

static from_stream(stream: ~typing.BinaryIO, kind: type[~runtimepy.primitives.base.Primitive[int]] = <class 'runtimepy.primitives.int.Uint16Primitive'>, byte_order: ~runtimepy.primitives.byte_order.ByteOrder = ByteOrder.NETWORK) → StringPrimitive[source]#: Create a new string primitive from a stream.

read(stream: BinaryIO) → str[source]#: Read a string from the stream.

set(value: str) → None[source]#: Set a new value for the underlying string.

property size: int#: Get the overall size of this string primitive.

property value: str#: Get the value of this string.

write(stream: BinaryIO) → int[source]#: Write this string’s size and value to the stream.

Module contents#

A module implementing a primitive-type storage entity.

class runtimepy.primitives.BaseFloatPrimitive(value: float = 0.0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: PrimitiveIsCloseMixin[float]

A simple primitive class for floating-point numbers.

randomize(timestamp_ns: int = None) → None[source]#: Set this primitive to a random integer.

async wait_for_value(value: float, timeout: float, operation: Operator = Operator.EQUAL) → EvalResult[source]#: Wait for this primitive to reach a specified state.

class runtimepy.primitives.BaseIntPrimitive(value: int = 0, scaling: list[float | int] = None, **kwargs)[source]#

Bases: PrimitiveIsCloseMixin[int]

A simple primitive class for integer primitives.

increment(amount: int = 1, timestamp_ns: int = None) → int[source]#: Increment this primitive by some amount and return the new value.

randomize(timestamp_ns: int = None) → None[source]#: Set this primitive to a random integer.

async wait_for_increment(timeout: float, count: int = 1) → EvalResult[source]#: Wait for this primitive to increment by a certain amount.

async wait_for_value(value: int | float, timeout: float, operation: Operator = Operator.EQUAL) → EvalResult[source]#: Wait for this primitive to reach a specified state.

runtimepy.primitives.Bool#: alias of BooleanPrimitive

runtimepy.primitives.Double#: alias of DoublePrimitive

runtimepy.primitives.Float#: alias of FloatPrimitive

runtimepy.primitives.Half#: alias of HalfPrimitive

runtimepy.primitives.Int16#: alias of Int16Primitive

runtimepy.primitives.Int32#: alias of Int32Primitive

runtimepy.primitives.Int64#: alias of Int64Primitive

runtimepy.primitives.Int8#: alias of Int8Primitive

class runtimepy.primitives.Primitive(value: ~runtimepy.primitives.base.T = None, scaling: list[float | int] = None, time_source: ~typing.Callable[[], int] = <function default_time_ns>)[source]#

Bases: Generic[T]

A simple class for storing an underlying primitive value.

age_ns(now: int = None) → int[source]#: Get the age of this primitive’s value in nanoseconds.

age_str(now: int = None) → str[source]#: Get the age of this primitive’s value as a string.

binary(byte_order: ByteOrder = None) → bytes[source]#: Convert this instance to a byte array.

byte_order: ByteOrder = 4#

callback(callback: Callable[[T, T], None]) → Iterator[None][source]#: Register a callback as a managed context.

callbacks: dict[int, tuple[Callable[[T, T], None], bool]]#

copy() → Primitive[T][source]#: A simple wrapper for copy.

curr_callback: int#

classmethod decode(data: bytes | bytearray | memoryview, byte_order: ByteOrder = None) → T[source]#: Decode a primitive of this type from provided data.

classmethod encode(value: T, byte_order: ByteOrder = None) → bytes[source]#: Create a bytes instance based on this primitive type.

from_stream(stream: BinaryIO, byte_order: ByteOrder = None) → T[source]#: Update this primitive from a stream and return the new value.

invert(value: T) → float | int[source]#: Invert a value using this primitive’s scaling.

kind: Int8Type | Int16Type | Int32Type | Int64Type | Uint8Type | Uint16Type | Uint32Type | Uint64Type | HalfType | FloatType | DoubleType | BooleanType#

last_updated_ns: int#

classmethod read(stream: BinaryIO, byte_order: ByteOrder = None) → T[source]#: Read a primitive from the provided stream based on this primitive type.

register_callback(callback: Callable[[T, T], None], once: bool = False) → int[source]#: Register a callback and return an identifier for it.

remove_callback(callback_id: int) → bool[source]#: Remove a callback if one is registered with this identifier.

scale(value: T) → float | int[source]#: Scale a value using this primitive’s scaling.

property scaled: float | int#: Get this primitive as a scaled value.

set_value(value: T, timestamp_ns: int = None) → None[source]#: Set a new underlying value.

property size: int#: Get the size of this primitive.

to_stream(stream: BinaryIO, byte_order: ByteOrder = None) → int[source]#: Write this primitive to a stream.

update(data: bytes | bytearray | memoryview, byte_order: ByteOrder = None) → T[source]#: Update this primitive from a bytes object.

classmethod valid_primitive(primitive: bool | int | float) → bool[source]#: Determine if a Python primitive is valid for this class.

property value: T#: Obtain the underlying value.

classmethod write(value: T, stream: BinaryIO, byte_order: ByteOrder = None) → int[source]#: Write a primitive to the stream based on this type.

class runtimepy.primitives.StrToBool(result: bool, valid: bool)[source]#

Bases: NamedTuple

A container for results when converting strings to boolean.

static check(data: str) → bool[source]#: Check a string for a boolean ‘true’ value.

static parse(data: str) → StrToBool[source]#: Parse a string to boolean.

result: bool#: Alias for field number 0

valid: bool#: Alias for field number 1

runtimepy.primitives.Uint16#: alias of Uint16Primitive

runtimepy.primitives.Uint32#: alias of Uint32Primitive

runtimepy.primitives.Uint64#: alias of Uint64Primitive

runtimepy.primitives.Uint8#: alias of Uint8Primitive

runtimepy.primitives.create(value: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str, **kwargs) → Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive[source]#: Create an instance of a primitive.

runtimepy.primitives.normalize(value: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str) → type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive][source]#: Normalize a type of primitive or a string into a type of primitive.

runtimepy.primitives package

Contents

runtimepy.primitives package#

Subpackages#

Submodules#

runtimepy.primitives.base module#

runtimepy.primitives.bool module#

runtimepy.primitives.byte_order module#

runtimepy.primitives.evaluation module#

runtimepy.primitives.float module#

runtimepy.primitives.int module#

runtimepy.primitives.scaling module#

runtimepy.primitives.string module#

Module contents#