QUnits

BaseUnit

class qrobot_qunits.base.BaseUnit(name: str, sampling_period: float | int, redis_config: RedisConfig | None = None, logging_config: LoggingConfig | None = None)

Base class for periodically scheduled, Redis-connected units.

Each unit runs _unit_task in a child process at sampling_period intervals and publishes its externally visible state under its unique ID.

Parameters:

• name (str) – Human-readable unit name used as the ID prefix.

• sampling_period (float) – Seconds between task executions. The minimum is MIN_TS.

id

Unique instance identifier composed from name and a random suffix.

Type:

str

name

Human-readable unit name.

Type:

str

sampling_period

Seconds between task executions.

Type:

float

start() → None

Start the unit’s background process and publish its type.

stop(timeout: float = 5.0) → None

Stop the worker and delete its Redis keys.

The worker finishes its current task before exiting. If it does not exit within timeout seconds, it is terminated as a last resort. Redis keys owned by the unit are removed after either a graceful or forced exit.

Parameters:

timeout (float) – Seconds to wait for a graceful exit before forcing termination.

QUnit

class qrobot_qunits.QUnit(name: str, model: Model, burst: Burst, sampling_period: float | int, query: list[float] | None = None, in_qunits: dict[int, str] | None = None, default_input: list[float] | None = None, redis_config: RedisConfig | None = None, logging_config: LoggingConfig | None = None)

Periodically process coupled inputs through a quantum-like model.

A qUnit reads its inputs from Redis, encodes them over the model’s temporal window, applies a query, and publishes the resulting burst output back to Redis.

Parameters:

• name (str) – Human-readable qUnit name.

• model (qrobot.models.Model) – Quantum-like model used to encode each temporal window.

• burst (qrobot.bursts.Burst) – Rule that converts a decoded state to a normalized output.

• sampling_period (float) – Seconds between input samples.

• query (list, optional) – Query target with one value per model dimension. Defaults to the all-zero vector.

• in_qunits (dict[int, str], optional) – Mapping from model dimensions to upstream unit IDs. Each mapped unit’s Redis output supplies that dimension.

• default_input (List[float]) – Default input vector of scalar values to use as default value when qUnit does not have an available one. Defaults to one zero per model dimension.

id

Unique qUnit instance identifier.

Type:

str

name

Human-readable qUnit name.

Type:

str

model

Model used to encode temporal windows.

Type:

qrobot.models.Model

burst

Rule used to convert decoded states to outputs.

Type:

qrobot.bursts.Burst

sampling_period

Seconds between input samples.

Type:

float

default_input

Default input vector of scalar values to use as default value when qunit does not have an available one

Type:

list[float]

get_burst_output() → float | None

Return the latest burst output published by the qUnit.

Returns:

The latest burst output written by the unit on the Redis database.

Return type:

float or None

property in_qunits: dict[int, str | None]

Return input unit IDs indexed by model dimension.

Returns:

Complete dimension mapping; unconnected dimensions map to None.

Return type:

dict

property input_vector: list[float]

Read the current input vector from input Redis outputs.

Returns:

One normalized value per model dimension. Missing input outputs use the corresponding default_input value.

Return type:

list

property query: list[float]

Return the current query target.

Returns:

Normalized target value for each model dimension.

Return type:

list

set_input(dim: int, input_id: str) → None

Connect a new input to the specified dimension to the qUnit.

Parameters:

• dim (int) – The input dimension index

• input_id (str) – The new input unit ID

SensorialUnit

class qrobot_qunits.SensorialUnit(name: str, sampling_period: float | int, default_input: float | None = None, redis_config: RedisConfig | None = None, logging_config: LoggingConfig | None = None)

Unit periodically sending normalized scalar readings.

Parameters:

• name (str) – Human-readable sensor name.

• sampling_period (float) – Seconds between Redis publications.

• default_input (float) – Initial scalar reading. Defaults to 0.0.

id

Unique sensor instance identifier.

Type:

str

name

Human-readable sensor name.

Type:

str

sampling_period

Seconds between Redis publications.

Type:

float

default_input

Default input for the scalar readings when the SensorialUnit does not have an available one

Type:

float

property scalar_reading: float

Current scalar reading.

ActuatorUnit

class qrobot_qunits.ActuatorUnit(name: str, in_qunits: list[str] | tuple[str, ...], sampling_period: float | int, threshold: float = 0.5, default_input: float = 0.0, redis_config: RedisConfig | None = None, logging_config: LoggingConfig | None = None)

Map the normalized sum of qUnit bursts to behavioral activation.

This implements the actuator interface: read one or more qUnit bursts, average them, and activate a behavioral routine when the normalized sum is strictly greater than a configured threshold. The unit’s output is stored in Redis as 0.0 (inhibited) or 1.0 (active), so a simulated or physical routine can consume it without being coupled to the qUnit worker.

Parameters:

• name (str) – The actuator name.

• in_qunits (list[str] | tuple[str, ...]) – Identifiers of the qUnits whose bursts drive the actuator.

• sampling_period (float) – How often to evaluate the latest qUnit bursts.

• threshold (float) – Activate only when the normalized sum is strictly greater than this value. Defaults to 0.5.

• default_input (float) – Value used for a qUnit that has not published yet. Defaults to 0.0.

name

The actuator name.

Type:

str

id

Unique identifier for this actuator unit.

Type:

str

threshold

Activation threshold (normalized to [0, 1]).

Type:

float

sampling_period

How often to evaluate the latest qUnit bursts.

Type:

float

default_input

Fallback value when a qUnit has not published yet.

Type:

float

activation_for(normalized_sum: float) → float

Return the thresholded activation for a normalized input sum.

get_activation() → float | None

Return the latest activation published by this actuator.

property in_qunits: dict[int, str]

Input qUnit identifiers indexed for network visualization.

property input_vector: list[float]

Latest burst values, using the configured fallback when absent.

property normalized_sum: float

Mean of the latest input bursts.

Redis integration

Redis communication and key protocol for qUnits.

class qrobot_qunits.redis.RedisAttribute(*values)

Attributes published for a Redis-connected unit.

class qrobot_qunits.redis.RedisConfig(host: str = 'localhost', port: int = 6379, database: int = 0)

Connection settings for a Redis database used by qUnits.

Parameters:

• host (str) – Redis server hostname. Defaults to "localhost".

• port (int) – Redis server port. Defaults to 6379.

• database (int) – Redis logical database number. Defaults to 0.

class qrobot_qunits.redis.RedisKey(unit_id: str, attribute: RedisAttribute)

Parsed qUnit Redis key.

Parameters:

• unit_id (str) – Unit identifier portion of the key.

• attribute (RedisAttribute) – Published attribute represented by the key.

exception qrobot_qunits.redis.RedisWriteError

Raised when a qUnit cannot persist its state to Redis.

qrobot_qunits.redis.build_redis_key(unit_id: str, attribute: RedisAttribute | str) → str

Build a validated <unit-id> <attribute> Redis key.

Parameters:

• unit_id (str) – Non-empty unit identifier without surrounding whitespace.

• attribute (RedisAttribute or str) – Protocol attribute name or enum member.

Returns:

Redis key formed by joining the unit ID and attribute with one space.

Return type:

str

Raises:

• TypeError – If unit_id is not a string.

• ValueError – If the unit ID is malformed or the attribute is unknown.

qrobot_qunits.redis.flush_redis(config: RedisConfig | None = None) → None

Remove every key from the configured Redis logical database.

Parameters:

config (RedisConfig | None) – Connection settings for the database to clear. The default is database 0 on the local Redis server.

Warning

This operation is destructive for the selected Redis logical database.

qrobot_qunits.redis.get_redis(config: RedisConfig | None = None) → Redis

Return a Redis client with decoded string responses.

Parameters:

config (RedisConfig | None) – Connection settings. When omitted, use the local default Redis server.

Returns:

A lazily connected redis-py client configured with decode_responses=True.

Return type:

redis.Redis

qrobot_qunits.redis.parse_redis_key(key: str) → RedisKey | None

Parse a protocol key, returning None for malformed or unknown keys.

Parameters:

key (str) – Candidate Redis key.

Returns:

Parsed unit ID and attribute, or None when the key is outside the protocol.

Return type:

RedisKey or None

qrobot_qunits.redis.redis_status(config: RedisConfig | None = None) → dict[str, str]

Return the current key/value status of a Redis database.

Parameters:

config (RedisConfig | None) – Connection settings for the database to inspect.

Returns:

Mapping of every existing key to its decoded string value. Keys deleted while scanning are omitted.

Return type:

dict[str, str]