robots_demo package

Submodules

robots_demo.squid_inference_realtime module

robots_demo.squid_inference_realtime.adapt_named_time_series(data, column_names, feature_indices)[source]

Adapt a named-column table to the checkpoint input layout.

Parameters:

data – Raw numeric table.
column_names – Column names from the input file.
feature_indices – Model feature positions in the checkpoint layout.

Returns:

Tuple (model_time_series, true_positions, layout_description); model_time_series is None when the named layout is incompatible.

robots_demo.squid_inference_realtime.adapt_numeric_time_series(data, feature_indices)[source]

Adapt a numeric table by inferring one of the supported layouts.

Supported layouts include full tables with time/x/y(/theta), compact feature-only tables, and compact tables with a leading time column.

Parameters:

data – Raw numeric table.
feature_indices – Model feature positions in the checkpoint layout.

Returns:

Tuple (model_time_series, true_positions, layout_description).

Raises:

ValueError – If the column count does not match any supported layout.

robots_demo.squid_inference_realtime.adapt_time_series_for_inference(data, column_names, inference_state)[source]

Adapt raw input data to the model layout used for inference.

Named-column detection is attempted first when names are available, then numeric-layout inference is used as a fallback.

Parameters:

data – Raw numeric table.
column_names – Optional input column names.
inference_state – Loaded checkpoint inference state.

Returns:

Tuple (model_time_series, true_positions, layout_description).

robots_demo.squid_inference_realtime.checkpoint_feature_indices(inference_state)[source]

Extract ordered input feature indices from a checkpoint state.

Parameters:: inference_state – State returned by load_center_cycle_for_inference.
Returns:: Ordered unique feature indices expected by the model.
Raises:: ValueError – If no feature indices are present.

robots_demo.squid_inference_realtime.data_row_to_file_row(path, data_row_index, first_data_file_row)[source]

Map a zero-based data row index to a user-facing file row number.

Parameters:

path – Input file path.
data_row_index – Zero-based row index in the loaded data array.
first_data_file_row – One-based first data row for text files.

Returns:

One-based row number for logging.

robots_demo.squid_inference_realtime.default_normalized_gif_file(gif_file)[source]

Create a default normalized-GIF path from the raw GIF path.

Parameters:: gif_file – Raw GIF output path.
Returns:: Path with _normalized before the extension.

robots_demo.squid_inference_realtime.file_signature(path)[source]

Return the modification-time and size signature for a file.

Parameters:: path – File path to inspect.
Returns:: Tuple (mtime_ns, size).

robots_demo.squid_inference_realtime.final_gif_update_message(args, prediction_count)[source]

Build the final log message for a last GIF update.

Parameters:

args – Parsed CLI namespace with GIF output settings.
prediction_count – Number of predictions written to the GIF.

Returns:

Final update message.

robots_demo.squid_inference_realtime.finite_true_positions_or_none(true_positions)[source]

Return finite true positions or None when unavailable.

Parameters:: true_positions – Optional true x,y position array.
Returns:: Float32 positions, or None if missing/all-NaN.

robots_demo.squid_inference_realtime.generated_time_values(row_count)[source]

Generate default monotonic time values for rows without timestamps.

Parameters:: row_count – Number of rows.
Returns:: Float32 array [0, 1, ..., row_count - 1].

robots_demo.squid_inference_realtime.gif_true_positions_or_none(predictions, true_positions)[source]

Return true positions only when aligned with predictions.

Parameters:

predictions – Prediction history.
true_positions – True-position history.

Returns:

Float32 true positions, or None if unusable.

robots_demo.squid_inference_realtime.gif_update_message(args, prediction_count)[source]

Build the inline log suffix for a GIF update.

Parameters:

args – Parsed CLI namespace with GIF output settings.
prediction_count – Number of predictions written to the GIF.

Returns:

Message suffix.

robots_demo.squid_inference_realtime.is_numeric_row(columns)[source]

Return whether a row contains at least one numeric value.

Empty columns are ignored.

Parameters:: columns – Iterable of column strings.
Returns:: True when every non-empty column parses as a float.

robots_demo.squid_inference_realtime.main()[source]

Parse CLI arguments and start realtime squid inference.

Raises:: ValueError – If CLI values are inconsistent or out of range.

robots_demo.squid_inference_realtime.make_expected_time_series(feature_values, time_values, feature_indices)[source]

Build the model input table expected by center-cycle inference.

Parameters:

feature_values – Compact feature matrix ordered like feature_indices.
time_values – Time column values.
feature_indices – Model feature positions in the checkpoint layout.

Returns:

Time column plus zero-filled model feature columns.

Raises:

ValueError – If the compact feature count does not match.

robots_demo.squid_inference_realtime.normalize_column_name(column_name)[source]

Normalize a column name for layout detection.

Parameters:: column_name – Raw column name.
Returns:: Lowercase stripped column name.

robots_demo.squid_inference_realtime.normalize_target_values(values, normalization_stats)[source]

Normalize target-space values with checkpoint statistics.

Parameters:

values – Values in original target units.
normalization_stats – Checkpoint normalization metadata.

Returns:

Normalized values.

robots_demo.squid_inference_realtime.normalized_target_rmse(prediction, true_position, normalization_stats)[source]

Compute RMSE in normalized target space.

Parameters:

prediction – Predicted target values in original units.
true_position – True target values in original units.
normalization_stats – Checkpoint normalization metadata.

Returns:

Normalized root mean squared error.

robots_demo.squid_inference_realtime.predict_window_targeting_row(time_series, target_row_index, inference_state, window_size)[source]

Predict the target position for a row using prior-window context.

Parameters:

time_series – Adapted model input table.
target_row_index – Row index whose prediction should be emitted.
inference_state – Loaded checkpoint inference state.
window_size – Number of prior rows required for the prediction.

Returns:

Predicted x,y values, or None when insufficient history is available.

robots_demo.squid_inference_realtime.read_text_time_series_file(path, delimiter, skip_header)[source]

Read a delimited text time-series file.

The first non-numeric line is treated as a header and skipped automatically.

Parameters:

path – Text file path.
delimiter – Optional delimiter passed to NumPy.
skip_header – Minimum number of header lines to skip.

Returns:

Tuple (data, column_names, first_data_file_row).

robots_demo.squid_inference_realtime.read_time_series_file(path, delimiter, skip_header, npz_key)[source]

Read a time-series table from text, .npy, or .npz input.

Parameters:

path – Input file path.
delimiter – Optional delimiter for text input.
skip_header – Header rows skipped for text input.
npz_key – Optional key for .npz archives.

Returns:

Tuple (data, column_names, first_data_file_row).

Raises:

ValueError – If the loaded data is not two-dimensional.

robots_demo.squid_inference_realtime.run_realtime_inference(args)[source]

Watch an input file and emit center-cycle predictions as rows arrive.

Parameters:: args – Parsed CLI namespace containing file, checkpoint, polling, window, GIF, and device settings.

robots_demo.squid_inference_realtime.save_normalized_realtime_gif(predictions, gif_file, interval_seconds, normalization_stats, true_positions=None)[source]

Normalize predictions and save a trajectory GIF.

Parameters:

predictions – Sequence of predicted positions in original units.
gif_file – Output GIF path.
interval_seconds – Frame interval for the animation.
normalization_stats – Checkpoint normalization metadata.
true_positions – Optional true positions in original units.

robots_demo.squid_inference_realtime.save_realtime_gif(predictions, gif_file, interval_seconds, true_positions=None)[source]

Write an animated trajectory GIF atomically.

Parameters:

predictions – Sequence of predicted x,y positions.
gif_file – Output GIF path.
interval_seconds – Frame interval for the animation.
true_positions – Optional true x,y positions to overlay.

robots_demo.squid_inference_realtime.save_realtime_gifs(predictions, args, normalization_stats, true_positions=None)[source]

Save raw and optionally normalized realtime trajectory GIFs.

Parameters:

predictions – Sequence of predicted positions.
args – Parsed CLI namespace with GIF output settings.
normalization_stats – Checkpoint normalization metadata.
true_positions – Optional true positions.

robots_demo.squid_inference_realtime.should_save_gif_for_batch(batch_row_count, batch_processed_rows, batch_prediction_count, save_interval)[source]

Return whether a GIF should be refreshed for a batch position.

Parameters:

batch_row_count – Number of newly available rows in the batch.
batch_processed_rows – Number of rows processed in the batch.
batch_prediction_count – Number of predictions emitted in the batch.
save_interval – Refresh interval for large batches.

Returns:

True when the GIF should be saved now.

robots_demo.squid_inference_realtime.split_column_line(line, delimiter)[source]

Split one header or data line into stripped columns.

Parameters:

line – Input text line.
delimiter – Optional delimiter; whitespace splitting is used when None.

Returns:

List of stripped column strings.

robots_demo.squid_inference_realtime.wait_until_file_settles(path, settle_seconds)[source]

Wait until a file signature remains unchanged across one interval.

Parameters:

path – File path to monitor.
settle_seconds – Seconds to wait between signature checks.

Returns:

Stable file signature.

robots_demo.train_cobot_classification module

robots_demo.train_cobot_classification.normalize_features(d, mean=None, std=None)[source]

Standardize sequence features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_cobot_classification.test(data_loader, classifier, scaler, label_encoder=None)[source]

Evaluate the cobot classifier on reservoir activations.

Parameters:

data_loader – Loader yielding normalized cobot sequences and labels.
classifier – Fitted classifier exposing predict.
scaler – Fitted scaler exposing transform.
label_encoder – Optional encoder used to decode labels before scoring.

Returns:

Tuple (accuracy, decoded_targets, decoded_predictions).

robots_demo.train_cobot_regression module

robots_demo.train_cobot_regression.normalize_features(d, mean=None, std=None)[source]

Standardize sequence features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_cobot_regression.normalize_targets(d, mean, std)[source]

Normalize regression targets with feature statistics.

Parameters:

d – Target array.
mean – Mean used for normalization.
std – Standard deviation used for normalization.

Returns:

Normalized target array.

robots_demo.train_cobot_regression.test(data_loader, regressor, scaler, state_average=False)[source]

Evaluate the cobot regressor on reservoir activations.

Parameters:

data_loader – Loader yielding normalized cobot sequences and targets.
regressor – Fitted regressor exposing predict.
scaler – Fitted scaler exposing transform.
state_average – If True, average hidden states over time.

Returns:

Root mean squared error.

robots_demo.train_squid module

robots_demo.train_squid.normalize_features(d, mean=None, std=None)[source]

Standardize squid sequence features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid.normalize_targets(d, mean=None, std=None)[source]

Standardize squid target values.

Parameters:

d – Target array.
mean – Optional target mean from the training split.
std – Optional target standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid.test(data_loader, regressor, scaler, state_average=False)[source]

Evaluate the squid regressor on reservoir activations.

Parameters:

data_loader – Loader yielding normalized squid sequences and targets.
regressor – Fitted regressor exposing predict.
scaler – Fitted scaler exposing transform.
state_average – If True, average hidden states over time.

Returns:

Root mean squared error.

robots_demo.train_squid_real module

robots_demo.train_squid_real.normalize_features(d, mean=None, std=None)[source]

Standardize real-squid sequence features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_real.normalize_targets(d, mean=None, std=None)[source]

Standardize real-squid target values.

Parameters:

d – Target array.
mean – Optional target mean from the training split.
std – Optional target standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_real.test(data_loader, regressor, scaler, state_average=False)[source]

Evaluate the real-squid regressor on reservoir activations.

Parameters:

data_loader – Loader yielding normalized real-squid sequences and targets.
regressor – Fitted regressor exposing predict.
scaler – Fitted scaler exposing transform.
state_average – If True, average hidden states over time.

Returns:

Root mean squared error.

robots_demo.train_squid_real_topology module

robots_demo.train_squid_real_topology.normalize_features(d, mean=None, std=None)[source]

Standardize real-squid topology features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_real_topology.normalize_targets(d, mean=None, std=None)[source]

Standardize real-squid topology target values.

Parameters:

d – Target array.
mean – Optional target mean from the training split.
std – Optional target standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_real_topology.test(data_loader, regressor, scaler, state_average=False)[source]

Evaluate the real-squid topology ensemble regressor.

Parameters:

data_loader – Loader yielding normalized sequences and targets.
regressor – Fitted regressor exposing predict.
scaler – Fitted scaler exposing transform.
state_average – If True, average hidden states over time.

Returns:

Root mean squared error.

robots_demo.train_squid_topology module

robots_demo.train_squid_topology.normalize_features(d, mean=None, std=None)[source]

Standardize squid topology features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_topology.normalize_targets(d, mean=None, std=None)[source]

Standardize squid topology target values.

Parameters:

d – Target array.
mean – Optional target mean from the training split.
std – Optional target standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_squid_topology.test(data_loader, regressor, scaler, state_average=False)[source]

Evaluate the squid topology ensemble regressor.

Parameters:

data_loader – Loader yielding normalized sequences and targets.
regressor – Fitted regressor exposing predict.
scaler – Fitted scaler exposing transform.
state_average – If True, average hidden states over time.

Returns:

Root mean squared error.

robots_demo.train_tomato module

robots_demo.train_tomato.normalize_features(d, mean=None, std=None)[source]

Standardize tomato sequence features across samples and timesteps.

Parameters:

d – Array with shape (n_samples, timesteps, n_features).
mean – Optional feature mean from the training split.
std – Optional feature standard deviation from the training split.

Returns:

Tuple (normalized, mean, std).

robots_demo.train_tomato.test(data_loader, classifier, scaler)[source]

Evaluate the tomato classifier on reservoir activations.

Parameters:

data_loader – Loader yielding normalized tomato sequences and labels.
classifier – Fitted classifier exposing score.
scaler – Fitted scaler exposing transform.

Returns:

Classification accuracy.

Module contents

Robot demonstration training and realtime inference scripts.