acies.vehicle_classifier.base#

Functions

`ensemble`(buff, win, size, conf_thresh)	Perform ensemble voting on the buffered predictions.
`get_twin_topic`(topic)	Convert between physical and digital twin topic.
`soft_vote`(pred_list)	Perform soft voting on a list of predictions.
`time_diff_decorator`(func)	Decorator to log the time difference between consecutive calls to a function.

Classes

Classifier

A template classifier for vehicle data.

class acies.vehicle_classifier.base.Classifier#

Bases: Service

A template classifier for vehicle data. This class should be subclassed and the load_model and infer methods implemented.

Parameters:

twin_buff_len (int)
sync_interval (int)
feature_twin (bool)

__init__(classifier_config_file, twin_model, twin_buff_len, sync_interval, feature_twin, *args, **kwargs)#

Initialize the Classifier.

Parameters:

classifier_config_file (str) – Path to the classifier configuration file.
twin_model (str) – The digital twin model to use.
twin_buff_len (int) – The buffer length for the digital twin.
sync_interval (int) – The synchronization interval for the digital twin.
feature_twin (bool) – Whether to use the digital twin features.

combine_meta(meta_data)#

Combine metadata from two topics.

Parameters:: meta_data (dict[str, dict[int, dict]]) – The metadata to combine.
Raises:: IndexError – If the topic format is invalid.
Returns:: The combined metadata.
Return type:: dict

get_keys_per_node(modalities)#

Get the keys for each node based on the modalities.

Parameters:: modalities (list[str]) – The list of modalities to consider.
Returns:: A dictionary mapping each node to its keys.
Return type:: dict[str, list[str]]

handle_message()#: Handle incoming messages from the message queue.

infer(samples)#

Perform inference on incoming samples.

This method must be implemented by concrete classifiers. The scheduler periodically invokes it with new data.

Parameters:

samples (dict[str, dict[int, numpy.ndarray]]) –

Nested mapping of time-series data.

Keys (str): Topic name, such as rs1/mic or rs1/geo.
Values (dict): Data received from that topic.
- Keys (int): Unix timestamp in seoncds.
- Values (numpy.ndarray): One-dimensional arrays containing the raw samples recorded during that second.

Example

samples = {
    "rs1-1/geo": {
        1757515699: array([16961, 16907, ..., 16551, 16370]),
        1757515700: array([16217, 16418, ..., 16913, 16799]),
    },
    "rs-1/mic": {
        1757515699: array([-62, -51, -63, ..., 182, 207, 178]),
        1757515700: array([194, 210, 174, ..., 233, 247, 210]),
    },
}

Returns:

Inference results produced by the classifier.

Return type:

Any

Raises:

NotImplementedError – If the subclass does not override this method.

load_model(*args, **kwargs)#: Load model from give path.

log_activate_status()#: Log the activation status.

run()#: Run the main loop.

temp_ensmeble(node, msg)#

Perform temporal ensemble on the classification results.

Parameters:

node (str) – The node name.
msg (Message) – The message containing the classification results.

Raises:

ValueError – If the ensemble buffer is not sufficient.

acies.vehicle_classifier.base.ensemble(buff, win, size, conf_thresh)#

Perform ensemble voting on the buffered predictions.

Parameters:

buff (EnsembleBuffer) – A buffer containing the predictions to ensemble.
win (int) – The time window (in seconds) to consider for ensembling.
size (int) – The minimum number of predictions required for ensembling.
conf_thresh (dict) – A dictionary of confidence thresholds for each class.

Raises:

ValueError – If there are not enough predictions for ensembling.

Returns:

The ensemble prediction and metadata.

Return type:

tuple[dict[str, float], dict]

acies.vehicle_classifier.base.get_twin_topic(topic)#

Convert between physical and digital twin topic.

Parameters:: topic (str) – The topic to convert.
Returns:: The converted topic.
Return type:: str

acies.vehicle_classifier.base.soft_vote(pred_list)#

Perform soft voting on a list of predictions.

Parameters:: pred_list (list[dict[str, float]]) – List of prediction dictionaries.
Returns:: A dictionary with the mean prediction scores.
Return type:: dict[str, float]

acies.vehicle_classifier.base.time_diff_decorator(func)#

Decorator to log the time difference between consecutive calls to a function.

Parameters:: func (Callable) – The function to decorate.
Returns:: The decorated function.
Return type:: Callable