API Reference
This section provides detailed API documentation for all classes and functions in ssl_simulator.
ssl_simulator
Top-level public API for ssl_simulator.
config
Configuration for ssl_simulator.
controllers
ConstantSignal
Oscillator
Bases: System
Write a 2-D velocity command tracking y = A*sin(w t) at constant speed s.
Time-driven (reads only world.time); writes a 2-D command component (default "u").
gamma is exposed for inspection.
constant_signal
Constant-signal control system.
ConstantSignal
oscillator
Oscillator control system -- follow y = A·sin(ω t) at constant speed.
Oscillator
Bases: System
Write a 2-D velocity command tracking y = A*sin(w t) at constant speed s.
Time-driven (reads only world.time); writes a 2-D command component (default "u").
gamma is exposed for inspection.
core
ssl_simulator core -- a data-oriented (ECS) simulation core.
Data lives in a :class:~ssl_simulator.core.world.World (struct-of-arrays component registry);
behaviour is :class:~ssl_simulator.core.system.System callables that mutate those arrays in place,
vectorized over the N entities; a :class:~ssl_simulator.core.scheduler orders systems by their
declared reads/writes; and :class:~ssl_simulator.core.engine.Engine ticks them and logs. See
docs/architecture.md.
Engine
__init__(time_step=0.01, log_filename=None, log_time_step=None, log_size=None, sink=None)
sink: optional callable(time, frame) streaming hook.
Called at the same cadence as the logger with the same snapshot {name: array} -- the
seam live viewers and telemetry publishers plug into (e.g. ssl-vista's
StreamSource.push, whose signature matches exactly). The frame holds live views of
the world's component arrays; a sink that retains frames must copy (StreamSource does).
IntegrationSystem
Bases: System
Advance integrated state by retraction: state ⊞ dt·deriv for each (state, deriv) pair.
The manifold is looked up per state component from world.manifolds (default: Euclidean), so
flat and Lie-group states run through the same loop. Runs last in the tick -- it writes the
loop-carried state consumed by the next tick's controllers.
Manifold
A state manifold with an in-place retraction x ⊞ ξ.
retract(x, tangent)
Advance x by tangent in place and return it.
Monitor
One tracked scalar with a running reduction and an optional one-shot warning.
observe(sample)
Fold sample into the running reduction and return it as a float.
tick(time)
Sample the source, update the reduction, and fire the warning once if tripped.
MonitorSystem
Bases: System
Tick every monitor once per step (observability as a scheduled pass).
A monitor is a tracked field with a running reduction and/or a one-shot warning (see
:mod:ssl_simulator.core.diagnostics); advancing that state is behaviour, so it is a system.
Runs last so it observes the fully computed step.
System
Base class for a scheduled system.
Subclasses set :attr:reads / :attr:writes (component names) and implement :meth:run.
Optional :attr:after / :attr:before (system names) break scheduling ties; :attr:monitors
lists diagnostics collected when the system is registered.
World
add(name, dim=None, *, init=None, dtype=float)
Register a component array of shape (N, *dim) (or seed it from init).
add_state(name, dim=None, *, init=None, manifold=None)
Register a component and mark it as integrated state on manifold (default flat).
add_system(system)
Register a system; collect declared observability/metadata.
A system may expose monitors (ticked diagnostics), tracked ({name: source}
scalars logged each step), and params ({name: value} static run metadata -> settings).
check()
Report dataflow problems -- unknown component names first (see :mod:.introspect).
describe()
Components, systems, and the dataflow between them (see :mod:.introspect).
track(name, source)
Register a global scalar source() to log each step (sampled at snapshot time).
build_schedule(systems, state_components)
Return systems in a valid execution order (see module docstring).
check_world(world)
Report dataflow problems: unknown names first, then suspicious wiring.
Catches the failure this architecture cannot raise on by itself - a system declaring a component that does not exist, which silently mis-orders the schedule.
describe_world(world)
A human-readable dump of the world: components, systems, and the dataflow between them.
map_entities(func, *arrays)
Apply func to each entity's slice across the leading axis and stack the results.
arrays are (N, ...) component arrays; func receives the i-th slice of each and
returns that entity's result. The stacked (N, ...) output can be written back in place
(world["x"][:] = map_entities(...)). This is the non-vectorized fallback - prefer a real
batched op where one exists.
batch
Running non-vectorized operations over the entity axis (ECS core).
Systems present a batched (N, ...) interface, but some math is only available one element at
a time - e.g. the lieplusplus Lie-group ops (single-element), or a scalar C/pybind kernel that
matches the Paparazzi autopilot. :func:map_entities is the fallback: it loops over the leading
(entity) axis, applies a single-entity function, and stacks the results, so a system can wrap a
scalar op while keeping its batched contract.
The rule of thumb: vectorize when the op supports it; fall back to :func:map_entities when it
doesn't. Because the system's (N, ...) interface is unchanged either way, swapping a scalar op
for a future batched implementation (a vectorized wheel, a GPU kernel) requires no change to the
system or the world - only the op behind the loop.
map_entities(func, *arrays)
Apply func to each entity's slice across the leading axis and stack the results.
arrays are (N, ...) component arrays; func receives the i-th slice of each and
returns that entity's result. The stacked (N, ...) output can be written back in place
(world["x"][:] = map_entities(...)). This is the non-vectorized fallback - prefer a real
batched op where one exists.
diagnostics
Observability primitive -- a running monitor (ECS core).
A :class:Monitor wraps a scalar source sampled every step: it reduces the samples over the
run (last/max/min/mean/sum) and optionally fires a one-shot warning the
first time a predicate goes true. A system that wants a diagnostic declares one and lists it in its
monitors; the :class:~ssl_simulator.core.system.MonitorSystem ticks them, and the engine logs
each monitor's reduced value and emits an end-of-run summary for any that tripped.
Example -- a running-max diagnostic that warns once on a threshold crossing::
self.monitors = [Monitor(
"max_step_disp", lambda: self._step_disp, reduce="max",
warn_if=lambda v: v > self.max_step,
warn_msg="cap engaged: raw step {value:.3g} > {name}",
)]
Monitor
One tracked scalar with a running reduction and an optional one-shot warning.
observe(sample)
Fold sample into the running reduction and return it as a float.
tick(time)
Sample the source, update the reduction, and fire the warning once if tripped.
engine
Engine -- ticks a World's scheduled systems and logs (ECS core).
The engine is deliberately thin: build the schedule once, run it each tick over the world's
component arrays, snapshot to the :class:~ssl_simulator.core.loggers.DataLogger at the log
cadence, and finalize diagnostics. All data lives in the :class:~ssl_simulator.core.world.World;
the engine owns only the clock and the logger.
Engine
__init__(time_step=0.01, log_filename=None, log_time_step=None, log_size=None, sink=None)
sink: optional callable(time, frame) streaming hook.
Called at the same cadence as the logger with the same snapshot {name: array} -- the
seam live viewers and telemetry publishers plug into (e.g. ssl-vista's
StreamSource.push, whose signature matches exactly). The frame holds live views of
the world's component arrays; a sink that retains frames must copy (StreamSource does).
introspect
Debugging tools -- inspect what a :class:~ssl_simulator.core.world.World actually contains.
In a data-oriented core, data and behaviour meet only through names. That is what makes the
model composable, but it also means a mistyped component in a system's reads/writes fails
silently: nothing raises, the scheduler simply orders that system wrong and the run produces
plausible-looking rubbish. These helpers make the wiring visible.
>>> print(world.describe()) # components, systems, and who touches what
>>> print(world.check()) # dataflow problems, typos first
Both are pure inspection - they never mutate the world.
check_world(world)
Report dataflow problems: unknown names first, then suspicious wiring.
Catches the failure this architecture cannot raise on by itself - a system declaring a component that does not exist, which silently mis-orders the schedule.
describe_world(world)
A human-readable dump of the world: components, systems, and the dataflow between them.
loggers
Compatibility shim: the streaming logger lives in the ground data plane now.
:class:DataLogger (and the canonical log formats it writes) moved to ssl_link.persistence
so every producer -- this simulator, a telemetry recorder, any future writer -- shares one
on-disk standard. Import from ssl_link in new code.
manifold
Manifolds -- the retraction primitive for integrated state (ECS core).
A :class:Manifold advances a state array by a tangent-space increment, in place. It is the
data-oriented form of the manifold ⊞ operation (JAX-MD's shift_fn): flat-space and
Lie-group states share one integration path, differing only in their retraction. The
:class:~ssl_simulator.core.system.IntegrationSystem calls manifold.retract(world[state],
dt * world[deriv]) -- this is the boundary the shared C++/pybind math (matching the Paparazzi
autopilot) plugs into.
Representations are flat struct-of-arrays over the N entities:
- :data:
Flat-- EuclideanR^d, state(N, d); retraction is addition (vectorized). - :data:
SO3-- rotations as(N, 3, 3)matrices; retractionR · exp(ξ)via lieplusplus (the same C++ Lie library the autopilot uses), through its array operators: one crossing into C++ per array rather than one per entity. It writes straight into the state array viaout=, so the retraction allocates nothing. Requires the optionallieplusplusdependency (thelieextra).
Older lieplusplus builds have no array operators; the per-entity fallback through
:func:~ssl_simulator.core.batch.map_entities is kept for those, and is roughly 60x slower at
N=100. The interface is identical either way, so no system can tell the difference.
Manifold
A state manifold with an in-place retraction x ⊞ ξ.
retract(x, tangent)
Advance x by tangent in place and return it.
scheduler
System scheduler -- orders systems by declared data dependencies (ECS core).
Within one tick a system that reads a component must run after the system that writes it. The
one subtlety is integrated state: it is loop-carried (available at tick start from the previous
tick's integration), so reading a state component creates no ordering edge -- otherwise the
control->dynamics->integration->(state) loop would look like a cycle. The IntegrationSystem is the
sole writer of state and naturally lands last.
Ordering is a stable topological sort (Kahn): among systems whose dependencies are satisfied, the
earliest-registered runs first, so independent systems keep registration order. after / before
hints (by system name) add explicit edges. A genuine non-state cycle raises ValueError.
build_schedule(systems, state_components)
Return systems in a valid execution order (see module docstring).
system
Systems -- behaviour over the world's component arrays (ECS core).
A system is a callable with declared data dependencies: the component arrays it reads and
writes. :func:ssl_simulator.core.scheduler.build_schedule uses those declarations to order
systems by within-tick producer->consumer flow. A system body is vectorized over all N entities
(one numpy / future-C++ call), mutating world[...] in place -- never a per-entity Python loop.
Controllers/policies (params in __init__), dynamics, and comm-graph builders are all systems.
Plain functions also work if they carry .reads / .writes attributes.
IntegrationSystem
Bases: System
Advance integrated state by retraction: state ⊞ dt·deriv for each (state, deriv) pair.
The manifold is looked up per state component from world.manifolds (default: Euclidean), so
flat and Lie-group states run through the same loop. Runs last in the tick -- it writes the
loop-carried state consumed by the next tick's controllers.
MonitorSystem
Bases: System
Tick every monitor once per step (observability as a scheduled pass).
A monitor is a tracked field with a running reduction and/or a one-shot warning (see
:mod:ssl_simulator.core.diagnostics); advancing that state is behaviour, so it is a system.
Runs last so it observes the fully computed step.
System
Base class for a scheduled system.
Subclasses set :attr:reads / :attr:writes (component names) and implement :meth:run.
Optional :attr:after / :attr:before (system names) break scheduling ties; :attr:monitors
lists diagnostics collected when the system is registered.
world
The World -- the single home of all simulation data (ECS core).
A :class:World is struct-of-arrays over N entities:
components-- named mutable arrays(N, *shape): state, control signals, derivatives, per-step scalars. Systems read/write these in place;world["p"]returns the live array.manifolds-- the retraction for each integrated state component (default: none = Euclidean).params-- static run metadata (gains, graphs, masks); Brax's "Model". Logged as settings.systems/monitors-- the registered behaviour and observability.time-- the clock, set by the engine each tick.
Data lives here, behaviour is in :mod:ssl_simulator.core.system, and they meet only
through named component arrays.
World
add(name, dim=None, *, init=None, dtype=float)
Register a component array of shape (N, *dim) (or seed it from init).
add_state(name, dim=None, *, init=None, manifold=None)
Register a component and mark it as integrated state on manifold (default flat).
add_system(system)
Register a system; collect declared observability/metadata.
A system may expose monitors (ticked diagnostics), tracked ({name: source}
scalars logged each step), and params ({name: value} static run metadata -> settings).
check()
Report dataflow problems -- unknown component names first (see :mod:.introspect).
describe()
Components, systems, and the dataflow between them (see :mod:.introspect).
track(name, source)
Register a global scalar source() to log each step (sampled at snapshot time).
exceptions
Custom exceptions for ssl_simulator.
Provides a clear hierarchy of exceptions for different error scenarios.
ConfigurationError
Bases: SSLSimulatorError
Raised when configuration is invalid or missing.
ControllerError
Bases: SSLSimulatorError
Raised when there's an error in controller setup or execution.
InitializationError
Bases: SSLSimulatorError
Raised when initialization of simulator, robot model, or controller fails.
RobotModelError
Bases: SSLSimulatorError
Raised when there's an error in robot model setup or execution.
SSLSimulatorError
Bases: Exception
Base exception class for all ssl_simulator errors.
ValidationError
Bases: SSLSimulatorError
Raised when data validation fails.
logging
ssl_simulator logging infrastructure.
Provides: - Custom log levels (DEBUG_VERBOSE) - Shared formatters - LoggerManager for centralized setup - Decorators - Utilities for standalone quick configuration
LoggerManager
Manages logging specifically for ssl_simulator.
Does NOT touch the root logger or other packages. Each framework package gets its own handler.
enable_third_party(package_names, level=logging.WARNING)
Optionally enable logging from third-party packages at a high level.
Examples
manager = LoggerManager() manager.setup() manager.enable_third_party(["PyQt5", "matplotlib"], level="WARNING")
set_format(format_type, packages=None)
Set format for specific packages.
set_level(level, packages=None)
Set level for specific packages.
setup(level=logging.INFO, format_type='compact', packages=None, handler=None, inline_max_len=None, inline_max_keys=None)
Configure logging for specific packages (default: ssl_simulator).
This configures ONLY these packages, leaving other packages silent.
Parameters
level : int | str Log level for the configured packages format_type : str Formatter preset (simple, compact, standard, detailed, json) packages : list[str], optional Package names to configure (default: ["ssl_simulator"]) handler : logging.Handler, optional Custom handler (default: StreamHandler to stdout) inline_max_len : int, optional Maximum line length for inline dict rendering. Edits CONFIG["LOG_INLINE_MAX_LEN"]. inline_max_keys : int, optional Maximum keys to keep dicts inline. Edits CONFIG["LOG_INLINE_MAX_KEYS"].
Examples
LoggerManager().setup(level="DEBUG")
Only ssl_simulator logs at DEBUG
suppress_package(package_name)
Suppress logging from a specific package.
normalize_level(level)
Convert string level names to logging level integers.
requires_log_level(logger, minimum_level)
Decorator: skip function if logger doesn't meet minimum level.
set_log_format(format_type='simple')
Set ssl_simulator logging format.
Examples
set_log_format("standard") set_log_format("detailed")
set_log_level(level=logging.INFO)
Set ssl_simulator logging level.
Examples
set_log_level("DEBUG") set_log_level(logging.INFO)
setup_logging(level=logging.INFO, format_type='standard', inline_max_len=None, inline_max_keys=None)
Configure the root logger with a project formatter. Idempotent.
warn_once(logger, key, msg, *args)
Log msg at WARNING level only the first time this key is seen.
Process-global dedup by key -- for "warn about this condition once per process"
situations (deprecations, one-off config issues). For per-run one-shot events
prefer a per-instance flag or a diagnostics Monitor, which resets each run.
decorators
Logging decorators and utilities.
requires_log_level(logger, minimum_level)
Decorator: skip function if logger doesn't meet minimum level.
formatters
Logging formatters used across all projects.
HumanFormatter
Bases: Formatter
Formatter that appends extra fields as an indented block.
Parameters
fmt : str
Standard logging format string, passed to logging.Formatter.
fancy : bool, default False
If True, render 1D and 2D ndarrays as aligned vectors and matrices
when the record is at DEBUG level. Plain text otherwise.
JSONFormatter
Bases: Formatter
Emits each record as a single JSON line, with extra fields merged in.
ndarray data previews are included only at DEBUG level; INFO and above keep just shape/dtype to avoid bloated production logs.
SafeJSONEncoder
Bases: JSONEncoder
JSON encoder that handles numpy and other common non-serializable types.
By default, ndarrays are summarized to shape/dtype only. Pass
include_preview=True to also include data previews.
levels
Custom logging levels used across all projects.
normalize_level(level)
Convert string level names to logging level integers.
manager
Centralized logging for ssl_simulator and dependent apps.
LoggerManager
Manages logging specifically for ssl_simulator.
Does NOT touch the root logger or other packages. Each framework package gets its own handler.
enable_third_party(package_names, level=logging.WARNING)
Optionally enable logging from third-party packages at a high level.
Examples
manager = LoggerManager() manager.setup() manager.enable_third_party(["PyQt5", "matplotlib"], level="WARNING")
set_format(format_type, packages=None)
Set format for specific packages.
set_level(level, packages=None)
Set level for specific packages.
setup(level=logging.INFO, format_type='compact', packages=None, handler=None, inline_max_len=None, inline_max_keys=None)
Configure logging for specific packages (default: ssl_simulator).
This configures ONLY these packages, leaving other packages silent.
Parameters
level : int | str Log level for the configured packages format_type : str Formatter preset (simple, compact, standard, detailed, json) packages : list[str], optional Package names to configure (default: ["ssl_simulator"]) handler : logging.Handler, optional Custom handler (default: StreamHandler to stdout) inline_max_len : int, optional Maximum line length for inline dict rendering. Edits CONFIG["LOG_INLINE_MAX_LEN"]. inline_max_keys : int, optional Maximum keys to keep dicts inline. Edits CONFIG["LOG_INLINE_MAX_KEYS"].
Examples
LoggerManager().setup(level="DEBUG")
Only ssl_simulator logs at DEBUG
suppress_package(package_name)
Suppress logging from a specific package.
utils
Convenience functions for logging setup. Use these for quick configuration without creating LoggerManager instance.
set_log_format(format_type='simple')
Set ssl_simulator logging format.
Examples
set_log_format("standard") set_log_format("detailed")
set_log_level(level=logging.INFO)
Set ssl_simulator logging level.
Examples
set_log_level("DEBUG") set_log_level(logging.INFO)
setup_logging(level=logging.INFO, format_type='standard', inline_max_len=None, inline_max_keys=None)
Configure the root logger with a project formatter. Idempotent.
warn_once(logger, key, msg, *args)
Log msg at WARNING level only the first time this key is seen.
Process-global dedup by key -- for "warn about this condition once per process"
situations (deprecations, one-off config issues). For per-run one-shot events
prefer a per-instance flag or a diagnostics Monitor, which resets each run.
math
Graph
An undirected graph over N agents with a live Laplacian.
Wraps :func:build_B / :func:build_L_from_B: set the edge list Z and it maintains the
incidence matrix B, Laplacian L (and Lb = L ⊗ I₂), and algebraic connectivity
lambda2. Agents can be dropped with :meth:kill_agents, which zeroes their edges.
gen_L()
Generate the Laplacian matrix considering agent status.
kill_agents(agents_index)
Kill the connections of the given agents and mark them non-active.
set_Z(Z)
Set the new Z and build the Laplacian matrix.
Q_prod_xi(Q, X)
Apply matrix Q to each row of X.
Parameters
Q : ndarray of shape (D, D) Transformation matrix. X : ndarray of shape (N, D) Input data where each row is a vector to be transformed.
Returns
X_transformed : ndarray of shape (N, D) Result of applying Q to each row of X.
Example
X = np.random.randn(10, 5) Q = np.eye(5) * 2 X2 = Q_prod_xi(Q, X) # Doubles each row
R_2D_matrix(angle)
Generate a 2D rotation matrix for a given angle.
Parameters
angle : float Rotation angle in radians.
Returns
np.ndarray
2x2 rotation matrix that rotates vectors counterclockwise by angle radians.
XY_distrib(N, rc0, lims, scale=1, n=2)
Generate a uniform distribution of points in a rectangular region.
Parameters
N : int Number of points to generate. rc0 : array-like of shape (n,) Central point in real space around which the points are distributed. lims : array-like of shape (n,) Range limits for each dimension (defines the half-length of the box in each direction). scale : float, optional Scaling factor applied to the overall distribution (default is 1). n : int, optional Number of dimensions (default is 2).
Returns
numpy.ndarray of shape (N, n) Array of generated points.
adapt_to_nd(X, target_ndim, dtype=None)
Adapt the input array to the specified number of dimensions.
Parameters
X : array-like
Input data to adapt.
target_ndim : int
Target number of dimensions.
dtype : data-type, optional
Desired data type of the output array.
Returns
np.ndarray
Array adapted to the specified number of dimensions.
angle_of_vectors(A, B)
Calculate the signed angle between pairs of 2D vectors A and B.
Parameters
A : numpy.ndarray of shape (N, 2) First set of 2D vectors. B : numpy.ndarray of shape (N, 2) Second set of 2D vectors.
Returns
theta : numpy.ndarray of shape (N,) Signed angles between each pair of vectors, in radians.
batman_distrib(N, rc0, lims, scale=1)
Batman distribution in 2D. * N: number of points * rc0: position in the real space of the central point * lims = [xlim, ylim]: width and length of the distribution.
build_B(list_edges, N)
Generate the incidence matrix for a graph.
Parameters
list_edges : list of tuple[int, int] List of edges, where each edge is represented as a tuple (tail, head). N : int Number of nodes in the graph.
Returns
np.ndarray Incidence matrix of shape (N, E), where E is the number of edges.
Note
This definition of the incidence matrix is for computing z = sum(x_i - x_j). If you need z = sum(x_j - x_i), multiply B by -1.
build_L_from_B(B, W=None)
Compute the Laplacian matrix from the incidence matrix.
The Laplacian matrix is defined as: L = B W B^T
Parameters
B : np.ndarray Incidence matrix of shape (N, E), where N is the number of nodes and E is the number of edges. W : np.ndarray, optional Diagonal weight matrix of shape (E, E), where each diagonal entry corresponds to the weight of an edge. If None, assumes uniform weights.
Returns
np.ndarray Laplacian matrix of shape (N, N).
Note
If no weight matrix W is provided, it defaults to the unweighted case: L = B B^T.
check_and_parse_dimensions(array, expected_shape, name=None, fill_values=None, dtype=float)
Generic function to check and parse dimensions of an array.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
array
|
ndarray
|
The input array to validate. |
required |
expected_shape
|
tuple
|
The expected shape of the array.
- Use |
required |
name
|
str
|
The name of the variable (for error messages). If None, attempts to infer the variable name. |
None
|
fill_values
|
int | list[int]
|
Value(s) to replace |
None
|
Returns
np.ndarray: The reshaped or validated array.
Raises
ValueError: If the array does not match the expected shape.
Examples
>>> arr = np.ones((10, 32, 64))
>>> check_and_parse_dimensions(arr, (None, 32, 64))
# passes, first dim is free (10)
>>> check_and_parse_dimensions(arr, (None, 32, 64), fill_values=10)
# passes only if first dim == 10
>>> arr2 = np.ones((5, 32, 7, 64))
>>> check_and_parse_dimensions(arr2, (None, 32, None, 64), fill_values=[5, 7])
# passes only if first==5 and third==7
>>> arr3 = np.ones((5, 2))
>>> check_and_parse_dimensions(arr3, (None, [2, 3]))
# passes, since second dim can be 2 or 3
>>> arr4 = np.ones((1, 3, 3))
>>> check_and_parse_dimensions(arr4, (5, 3, 3), fill_values=5).shape
# (5, 3, 3) -> broadcasted from (1, 3, 3)
circular_distrib(N, rc0=0, r=1, **kwargs)
Generate points approximately uniformly distributed in a circular annulus.
Parameters
N : int Number of points to generate. rc0 : array-like Center of the circular distribution. r : float Outer radius.
Returns
np.ndarray Array of shape (N, 2) containing the 2D coordinates of sampled points.
compose(a, b)
Compose poses: a @ b (apply b first, then a).
cov_matrix(X, sample=False)
Compute covariance matrix/matrices for D-dimensional vectors.
Supports both single and batched inputs: - Input shape (N, D): a single dataset of N samples in D dimensions. - Input shape (K, N, D): K datasets, each with N samples in D dimensions. Returns K covariance matrices.
Parameters
X : np.ndarray Input array of shape (N, D) or (K, N, D). sample : bool, default=False If True, compute the sample covariance (divide by N-1). If False, compute the population covariance (divide by N).
Returns
cov : np.ndarray - If input shape is (N, D), returns array of shape (D, D). - If input shape is (K, N, D), returns array of shape (K, D, D).
elliptical_distrib(N, rc0=0, rx=1.0, ry=1.0, h=0.0, border_noise=0.0, rot_angle=0.0)
Generate a uniform distribution in an elliptical (or circular) annulus.
Parameters
N : int Number of points rc0 : array-like Center of the ellipse rx : float Horizontal radius (semi-major axis) ry : float Vertical radius (semi-minor axis) h : float Inner radius (for annulus); use 0 for full disk border_noise : float Adds random jitter to the radial position
Returns
np.ndarray Array of shape (N, 2) with sampled positions
exp(X, Q, x0)
Compute the exponential of a quadratic form:
exp(X) = exp((X - x0)^T @ Q @ (X - x0))
Parameters
X : array-like of shape (N, D) Input points where the function is evaluated. Q : ndarray of shape (D, D) Quadratic form matrix. x0 : array-like of shape (D,) Center of the Gaussian (mean).
Returns
result : ndarray of shape (N,) Result of applying the exponential quadratic form to each point in X.
flower_formation(N, R, b=3)
Function to generate a non-uniform (dummy) "flower" distribution of N agents.
from_frame(points, pose)
Express frame-local points in world coordinates (apply the pose).
gen_Z_distance(P, dist_thr)
Generate a graph based on a distance threshold heuristic.
For each pair of nodes (i, j), if the distance between them (d_ij) is less than or equal to the given threshold (dist_thr), an edge (i, j) is added to the graph.
Parameters
P : np.ndarray An array of shape (N, D), where N is the number of nodes and D is the number of dimensions. Each row represents the coordinates of a node in the space. dist_thr : float The distance threshold. If the distance between two nodes is less than or equal to this value, an edge will be created between them.
Returns
list of tuple[int, int] List of edges, each represented as a tuple (i, j), where i and j are node indices.
gen_Z_random(N, rounds=1, seed=None)
Generate a random connected undirected graph using a heuristic.
This function ensures that the generated graph is connected by iteratively selecting edges between visited and non-visited nodes.
Parameters
N : int Number of nodes in the graph. rounds : int, optional Number of times to attempt adding additional edges (default is 1). seed : int, optional Random seed for reproducibility (default is None).
Returns
list of tuple[int, int] List of edges, represented as a tuple (tail, head), forming a connected graph.
gen_Z_ring(N)
Generate a ring-shaped graph where each node is connected to its two neighbors.
The graph consists of N nodes arranged in a ring structure, where node i is connected to node i+1, and node N-1 is connected to node 0 to form the ring.
Parameters
N : int Number of nodes in the graph.
Returns
list of tuple[int, int] List of edges, where each edge is represented as a tuple (i, j), forming a closed ring graph.
gen_Z_split(N, order, n_breaks=0)
Split a fully connected graph into n_breaks smaller fully connected graphs.
The graph is initially divided into order subgraphs. Afterward, a number of edges
(based on n_breaks) are removed from the generated graph to split it into disconnected subgraphs.
Parameters
N : int Number of nodes in the graph. order : int Number of subgraphs to create. n_breaks : int, optional Number of connections to remove in each subgraph (default is 0, meaning no edges are removed).
Returns
list of tuple[int, int] List of edges, represented as a tuple (i, j), forming the generated graph with subgraphs.
invert(pose)
Invert poses. T⁻¹ has rotation Rᵀ and translation -Rᵀt (never a generic solve).
make_pose(rotation, translation)
Assemble (…, 4, 4) poses from (…, 3, 3) rotations and (…, 3) translations.
norm_2(A)
Compute the matrix 2-norm (spectral norm) of a matrix A.
The 2-norm of a matrix is defined as the largest singular value of A, which is equivalent to the square root of the largest eigenvalue of AᵀA.
Parameters
A : np.ndarray Input matrix of shape (m, n).
Returns
float The spectral (operator 2-) norm of the matrix A.
pose_parts(pose)
Split (…, 4, 4) poses into (rotation, translation).
regpoly_formation(N, r, thetha0=0)
Function to generate a regular polygon distribution.
rigid_transform(src, dst)
Kabsch/Procrustes fit: the pose T minimising ‖T·src_i - dst_i‖.
A proper rigid motion - rotation only (det = +1, reflections corrected) and a translation,
never scaling. Needs at least 3 non-degenerate point pairs in 3-D.
This is the analysis counterpart to the frame changes above: given a point set in two frames it recovers the transform between them, which is how you check which frame an estimator actually converged to when no pose is known a priori.
rot_3d_matrix(roll, pitch, yaw, dec=None)
Generate R ∈ SO(3) from ROLL, PITCH, YAW. Fast for scalar inputs, supports arrays with broadcasting.
so3_hat(omega)
- Generate \omega_\hat ∈ so(3) from the \omega vector - Supports single vector (3,) or batch (N,3).
so3_vee(omega_hat)
- Generate \omega vector from \omega_\hat ∈ so(3) - Supports batch (...,3,3).
to_frame(points, pose)
Express world points in the frame's local coordinates (apply the inverse pose).
uniform_distrib(N, lims, rc0=None, seed=None)
Generate a uniform distribution of points within a hyper-rectangular region in arbitrary dimensions.
Parameters
N : int Number of points to generate. lims : list of float Distance limits [lim_1, lim_2, ..., lim_D] defining half-size of the box along each dimension. Total side length is 2 * lim_i. rc0 : list of float, optional Coordinates [c_1, c_2, ..., c_D] of the centroid of the distribution. If None, defaults to the origin (0,...,0). seed : int, optional Random seed for reproducibility.
Returns
np.ndarray Array of shape (N, D) containing the generated points.
Raises
ValueError
If rc0 or lims lengths do not match the dimension D.
unit_vec(V, delta=0, axis=-1)
Normalize a bundle of 2D vectors.
Parameters
V : np.ndarray
Input array of shape (..., 2), e.g. (T, N, 2).
delta : float
Threshold below which vectors are considered zero.
axis : int
Axis along which to normalize.
Returns
np.ndarray
Array of unit vectors, same shape as V.
algebra
R_2D_matrix(angle)
Generate a 2D rotation matrix for a given angle.
Parameters
angle : float Rotation angle in radians.
Returns
np.ndarray
2x2 rotation matrix that rotates vectors counterclockwise by angle radians.
cov_matrix(X, sample=False)
Compute covariance matrix/matrices for D-dimensional vectors.
Supports both single and batched inputs: - Input shape (N, D): a single dataset of N samples in D dimensions. - Input shape (K, N, D): K datasets, each with N samples in D dimensions. Returns K covariance matrices.
Parameters
X : np.ndarray Input array of shape (N, D) or (K, N, D). sample : bool, default=False If True, compute the sample covariance (divide by N-1). If False, compute the population covariance (divide by N).
Returns
cov : np.ndarray - If input shape is (N, D), returns array of shape (D, D). - If input shape is (K, N, D), returns array of shape (K, D, D).
norm_2(A)
Compute the matrix 2-norm (spectral norm) of a matrix A.
The 2-norm of a matrix is defined as the largest singular value of A, which is equivalent to the square root of the largest eigenvalue of AᵀA.
Parameters
A : np.ndarray Input matrix of shape (m, n).
Returns
float The spectral (operator 2-) norm of the matrix A.
basics
Q_prod_xi(Q, X)
Apply matrix Q to each row of X.
Parameters
Q : ndarray of shape (D, D) Transformation matrix. X : ndarray of shape (N, D) Input data where each row is a vector to be transformed.
Returns
X_transformed : ndarray of shape (N, D) Result of applying Q to each row of X.
Example
X = np.random.randn(10, 5) Q = np.eye(5) * 2 X2 = Q_prod_xi(Q, X) # Doubles each row
adapt_to_nd(X, target_ndim, dtype=None)
Adapt the input array to the specified number of dimensions.
Parameters
X : array-like
Input data to adapt.
target_ndim : int
Target number of dimensions.
dtype : data-type, optional
Desired data type of the output array.
Returns
np.ndarray
Array adapted to the specified number of dimensions.
angle_of_vectors(A, B)
Calculate the signed angle between pairs of 2D vectors A and B.
Parameters
A : numpy.ndarray of shape (N, 2) First set of 2D vectors. B : numpy.ndarray of shape (N, 2) Second set of 2D vectors.
Returns
theta : numpy.ndarray of shape (N,) Signed angles between each pair of vectors, in radians.
check_and_parse_dimensions(array, expected_shape, name=None, fill_values=None, dtype=float)
Generic function to check and parse dimensions of an array.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
array
|
ndarray
|
The input array to validate. |
required |
expected_shape
|
tuple
|
The expected shape of the array.
- Use |
required |
name
|
str
|
The name of the variable (for error messages). If None, attempts to infer the variable name. |
None
|
fill_values
|
int | list[int]
|
Value(s) to replace |
None
|
Returns
np.ndarray: The reshaped or validated array.
Raises
ValueError: If the array does not match the expected shape.
Examples
>>> arr = np.ones((10, 32, 64))
>>> check_and_parse_dimensions(arr, (None, 32, 64))
# passes, first dim is free (10)
>>> check_and_parse_dimensions(arr, (None, 32, 64), fill_values=10)
# passes only if first dim == 10
>>> arr2 = np.ones((5, 32, 7, 64))
>>> check_and_parse_dimensions(arr2, (None, 32, None, 64), fill_values=[5, 7])
# passes only if first==5 and third==7
>>> arr3 = np.ones((5, 2))
>>> check_and_parse_dimensions(arr3, (None, [2, 3]))
# passes, since second dim can be 2 or 3
>>> arr4 = np.ones((1, 3, 3))
>>> check_and_parse_dimensions(arr4, (5, 3, 3), fill_values=5).shape
# (5, 3, 3) -> broadcasted from (1, 3, 3)
exp(X, Q, x0)
Compute the exponential of a quadratic form:
exp(X) = exp((X - x0)^T @ Q @ (X - x0))
Parameters
X : array-like of shape (N, D) Input points where the function is evaluated. Q : ndarray of shape (D, D) Quadratic form matrix. x0 : array-like of shape (D,) Center of the Gaussian (mean).
Returns
result : ndarray of shape (N,) Result of applying the exponential quadratic form to each point in X.
unit_vec(V, delta=0, axis=-1)
Normalize a bundle of 2D vectors.
Parameters
V : np.ndarray
Input array of shape (..., 2), e.g. (T, N, 2).
delta : float
Threshold below which vectors are considered zero.
axis : int
Axis along which to normalize.
Returns
np.ndarray
Array of unit vectors, same shape as V.
distributions
XY_distrib(N, rc0, lims, scale=1, n=2)
Generate a uniform distribution of points in a rectangular region.
Parameters
N : int Number of points to generate. rc0 : array-like of shape (n,) Central point in real space around which the points are distributed. lims : array-like of shape (n,) Range limits for each dimension (defines the half-length of the box in each direction). scale : float, optional Scaling factor applied to the overall distribution (default is 1). n : int, optional Number of dimensions (default is 2).
Returns
numpy.ndarray of shape (N, n) Array of generated points.
batman_distrib(N, rc0, lims, scale=1)
Batman distribution in 2D. * N: number of points * rc0: position in the real space of the central point * lims = [xlim, ylim]: width and length of the distribution.
circular_distrib(N, rc0=0, r=1, **kwargs)
Generate points approximately uniformly distributed in a circular annulus.
Parameters
N : int Number of points to generate. rc0 : array-like Center of the circular distribution. r : float Outer radius.
Returns
np.ndarray Array of shape (N, 2) containing the 2D coordinates of sampled points.
elliptical_distrib(N, rc0=0, rx=1.0, ry=1.0, h=0.0, border_noise=0.0, rot_angle=0.0)
Generate a uniform distribution in an elliptical (or circular) annulus.
Parameters
N : int Number of points rc0 : array-like Center of the ellipse rx : float Horizontal radius (semi-major axis) ry : float Vertical radius (semi-minor axis) h : float Inner radius (for annulus); use 0 for full disk border_noise : float Adds random jitter to the radial position
Returns
np.ndarray Array of shape (N, 2) with sampled positions
flower_formation(N, R, b=3)
Function to generate a non-uniform (dummy) "flower" distribution of N agents.
regpoly_formation(N, r, thetha0=0)
Function to generate a regular polygon distribution.
uniform_distrib(N, lims, rc0=None, seed=None)
Generate a uniform distribution of points within a hyper-rectangular region in arbitrary dimensions.
Parameters
N : int Number of points to generate. lims : list of float Distance limits [lim_1, lim_2, ..., lim_D] defining half-size of the box along each dimension. Total side length is 2 * lim_i. rc0 : list of float, optional Coordinates [c_1, c_2, ..., c_D] of the centroid of the distribution. If None, defaults to the origin (0,...,0). seed : int, optional Random seed for reproducibility.
Returns
np.ndarray Array of shape (N, D) containing the generated points.
Raises
ValueError
If rc0 or lims lengths do not match the dimension D.
graphs
Graph
An undirected graph over N agents with a live Laplacian.
Wraps :func:build_B / :func:build_L_from_B: set the edge list Z and it maintains the
incidence matrix B, Laplacian L (and Lb = L ⊗ I₂), and algebraic connectivity
lambda2. Agents can be dropped with :meth:kill_agents, which zeroes their edges.
gen_L()
Generate the Laplacian matrix considering agent status.
kill_agents(agents_index)
Kill the connections of the given agents and mark them non-active.
set_Z(Z)
Set the new Z and build the Laplacian matrix.
build_B(list_edges, N)
Generate the incidence matrix for a graph.
Parameters
list_edges : list of tuple[int, int] List of edges, where each edge is represented as a tuple (tail, head). N : int Number of nodes in the graph.
Returns
np.ndarray Incidence matrix of shape (N, E), where E is the number of edges.
Note
This definition of the incidence matrix is for computing z = sum(x_i - x_j). If you need z = sum(x_j - x_i), multiply B by -1.
build_L_from_B(B, W=None)
Compute the Laplacian matrix from the incidence matrix.
The Laplacian matrix is defined as: L = B W B^T
Parameters
B : np.ndarray Incidence matrix of shape (N, E), where N is the number of nodes and E is the number of edges. W : np.ndarray, optional Diagonal weight matrix of shape (E, E), where each diagonal entry corresponds to the weight of an edge. If None, assumes uniform weights.
Returns
np.ndarray Laplacian matrix of shape (N, N).
Note
If no weight matrix W is provided, it defaults to the unweighted case: L = B B^T.
gen_Z_distance(P, dist_thr)
Generate a graph based on a distance threshold heuristic.
For each pair of nodes (i, j), if the distance between them (d_ij) is less than or equal to the given threshold (dist_thr), an edge (i, j) is added to the graph.
Parameters
P : np.ndarray An array of shape (N, D), where N is the number of nodes and D is the number of dimensions. Each row represents the coordinates of a node in the space. dist_thr : float The distance threshold. If the distance between two nodes is less than or equal to this value, an edge will be created between them.
Returns
list of tuple[int, int] List of edges, each represented as a tuple (i, j), where i and j are node indices.
gen_Z_random(N, rounds=1, seed=None)
Generate a random connected undirected graph using a heuristic.
This function ensures that the generated graph is connected by iteratively selecting edges between visited and non-visited nodes.
Parameters
N : int Number of nodes in the graph. rounds : int, optional Number of times to attempt adding additional edges (default is 1). seed : int, optional Random seed for reproducibility (default is None).
Returns
list of tuple[int, int] List of edges, represented as a tuple (tail, head), forming a connected graph.
gen_Z_ring(N)
Generate a ring-shaped graph where each node is connected to its two neighbors.
The graph consists of N nodes arranged in a ring structure, where node i is connected to node i+1, and node N-1 is connected to node 0 to form the ring.
Parameters
N : int Number of nodes in the graph.
Returns
list of tuple[int, int] List of edges, where each edge is represented as a tuple (i, j), forming a closed ring graph.
gen_Z_split(N, order, n_breaks=0)
Split a fully connected graph into n_breaks smaller fully connected graphs.
The graph is initially divided into order subgraphs. Afterward, a number of edges
(based on n_breaks) are removed from the generated graph to split it into disconnected subgraphs.
Parameters
N : int Number of nodes in the graph. order : int Number of subgraphs to create. n_breaks : int, optional Number of connections to remove in each subgraph (default is 0, meaning no edges are removed).
Returns
list of tuple[int, int] List of edges, represented as a tuple (i, j), forming the generated graph with subgraphs.
gvf
Guiding vector field (GVF) trajectories - implicit-function path representations.
gvf_ellipse
gvf_line
gvf_line_AB
gvf_line_heading
lie
This module provides functions for working with 3D rotations and the Lie algebra so(3). It includes utilities for generating rotation matrices, constructing orthonormal bases, and computing exponential and logarithmic maps between SO(3) and its Lie algebra.
Notes
- The module assumes input vectors and matrices are NumPy arrays.
- Some functions use approximations for small angles to improve numerical stability.
rot_3d_matrix(roll, pitch, yaw, dec=None)
Generate R ∈ SO(3) from ROLL, PITCH, YAW. Fast for scalar inputs, supports arrays with broadcasting.
so3_hat(omega)
- Generate \omega_\hat ∈ so(3) from the \omega vector - Supports single vector (3,) or batch (N,3).
so3_vee(omega_hat)
- Generate \omega vector from \omega_\hat ∈ so(3) - Supports batch (...,3,3).
scalar_fields
Scalar fields - field math for source-seeking (value / gradient / Hessian).
SigmaFract
Bases: ScalarField
Fractal scalar field.
Attributes
k: float
norm factor
mu: list
center of the Gaussian.
dev: float
models the scale of the distribution while maintaining its properties
a: np.ndarray
center of the first Gaussian
b: np.ndarray
center of the second Gaussian
Qa: numpy array
2x2 matrix, quadratic transformation of the first Gaussian input
Qb: numpy array
2x2 matrix, quadratic transformation of the second Gaussian input
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
get_config()
Returns the key parameters used for reinitialization.
SigmaGauss
Bases: ScalarField
Gaussian scalar field function.
Attributes
mu: list
center of the Gaussian.
max_intensity: float
scalar field value at the source
dev: float
models the width of the Gaussian
S: numpy array
2x2 matrix, rotation matrix applied to the scalar field
R: numpy array
2x2 matrix, scaling matrix applied to the scalar field
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
get_config()
Returns the key parameters used for reinitialization.
SigmaNonconvex
Bases: ScalarField
Non-convex scalar field function (two Gaussians + "norm factor" * norm).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
k
|
float norm factor |
required | |
mu
|
list center of the Gaussian. |
None
|
|
dev
|
float models the scale of the distribution while maintaining its properties |
1
|
|
a
|
np.ndarray center of the first Gaussian |
default_a
|
|
b
|
np.ndarray center of the second Gaussian |
default_b
|
|
Qa
|
numpy array 2x2 matrix, quadratic transformation of the first Gaussian input |
None
|
|
Qb
|
numpy array 2x2 matrix, quadratic transformation of the second Gaussian input |
None
|
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
L_sigma(X, sigma)
Cetralised L_sigma calculation function (only for numerical validation).
Attributes
X: numpy array
(N x 2) matrix of agents position from the centroid
sigma: numpy array
(N) vector of simgma_values on each row of X
calc_mu_centralized(X, sigma)
Cetralised mu calculation function (only for numerical validation).
Attributes
X: numpy array
(N x 2) matrix of agents position from the centroid
sigma: numpy array
(N) vector of simgma_values on each row of X
sigma_fract
Fractal function (two Gaussians + contant * norm).
SigmaFract
Bases: ScalarField
Fractal scalar field.
Attributes
k: float
norm factor
mu: list
center of the Gaussian.
dev: float
models the scale of the distribution while maintaining its properties
a: np.ndarray
center of the first Gaussian
b: np.ndarray
center of the second Gaussian
Qa: numpy array
2x2 matrix, quadratic transformation of the first Gaussian input
Qb: numpy array
2x2 matrix, quadratic transformation of the second Gaussian input
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
get_config()
Returns the key parameters used for reinitialization.
create_Qa()
Create the quadratic transformation matrix Qa for the first Gaussian.
create_Qb()
Create the quadratic transformation matrix Qb for the second Gaussian.
sigma_gauss
Gaussian function.
SigmaGauss
Bases: ScalarField
Gaussian scalar field function.
Attributes
mu: list
center of the Gaussian.
max_intensity: float
scalar field value at the source
dev: float
models the width of the Gaussian
S: numpy array
2x2 matrix, rotation matrix applied to the scalar field
R: numpy array
2x2 matrix, scaling matrix applied to the scalar field
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
get_config()
Returns the key parameters used for reinitialization.
sigma_nonconvex
Non-convex function (two Gaussians + contant * norm).
SigmaNonconvex
Bases: ScalarField
Non-convex scalar field function (two Gaussians + "norm factor" * norm).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
k
|
float norm factor |
required | |
mu
|
list center of the Gaussian. |
None
|
|
dev
|
float models the scale of the distribution while maintaining its properties |
1
|
|
a
|
np.ndarray center of the first Gaussian |
default_a
|
|
b
|
np.ndarray center of the second Gaussian |
default_b
|
|
Qa
|
numpy array 2x2 matrix, quadratic transformation of the first Gaussian input |
None
|
|
Qb
|
numpy array 2x2 matrix, quadratic transformation of the second Gaussian input |
None
|
L1(pc, P)
Function for calculating L^1.
Attributes
pc: numpy array [x,y] position of the centroid P: numpy array (N x 2) matrix of agents position
source_seeking
L_sigma(X, sigma)
Cetralised L_sigma calculation function (only for numerical validation).
Attributes
X: numpy array
(N x 2) matrix of agents position from the centroid
sigma: numpy array
(N) vector of simgma_values on each row of X
calc_mu_centralized(X, sigma)
Cetralised mu calculation function (only for numerical validation).
Attributes
X: numpy array
(N x 2) matrix of agents position from the centroid
sigma: numpy array
(N) vector of simgma_values on each row of X
transforms
Rigid pose transformations -- SE(3) poses and the frame changes built on them.
A pose here is a 4x4 homogeneous matrix T shaped (4, 4) (one pose) or (N, 4, 4)
(one per entity). The convention throughout is that a pose is the frame expressed in world, so
it maps frame-local coordinates to world coordinates:
``from_frame(p_local, T) -> p_world`` apply ``T``
``to_frame(p_world, T) -> p_local`` apply ``T⁻¹``
Points are (N, 3). A single (4, 4) pose broadcasts over all of them - the common case of
one shared reference frame - while an (N, 4, 4) stack pairs one pose per point, which is what
per-entity frames need.
Everything here is rigid-motion algebra, which numpy does well, so the module has no required
dependencies. When lieplusplus is available (the shared C++/pybind library the Paparazzi
autopilot uses) the work routes through its array operators instead - identical maths, several
times faster, and the same code the autopilot runs. They accept any layout or dtype and convert
internally, so no ascontiguousarray dance is needed here.
Note this module deliberately stops at frame algebra. Integrating a pose over time is a
manifold retraction, which lives with the other integrated-state manifolds in
:mod:ssl_simulator.core.manifold next to :data:~ssl_simulator.core.manifold.Flat and
:data:~ssl_simulator.core.manifold.SO3 - add an SE3 manifold there (lpp.se3_retract)
when a world needs an integrated pose state.
compose(a, b)
Compose poses: a @ b (apply b first, then a).
from_frame(points, pose)
Express frame-local points in world coordinates (apply the pose).
invert(pose)
Invert poses. T⁻¹ has rotation Rᵀ and translation -Rᵀt (never a generic solve).
make_pose(rotation, translation)
Assemble (…, 4, 4) poses from (…, 3, 3) rotations and (…, 3) translations.
pose_parts(pose)
Split (…, 4, 4) poses into (rotation, translation).
rigid_transform(src, dst)
Kabsch/Procrustes fit: the pose T minimising ‖T·src_i - dst_i‖.
A proper rigid motion - rotation only (det = +1, reflections corrected) and a translation,
never scaling. Needs at least 3 non-degenerate point pairs in 3-D.
This is the analysis counterpart to the frame changes above: given a point set in two frames it recovers the transform between them, which is how you check which frame an estimator actually converged to when no pose is known a priori.
to_frame(points, pose)
Express world points in the frame's local coordinates (apply the inverse pose).
robot_models
UnicycleDynamics
single_integrator
Single-integrator dynamics -- ẋ = u.
In the ECS core a single integrator needs no dynamics system: the command is the derivative, so
integration retracts the state directly by the command. This helper registers the state + command
components on a world and returns the :class:IntegrationSystem that advances them.
single_integrator(world, state='p', cmd='u', *, dim=None, init=None, manifold=Flat)
Register state (integrated on manifold) and its cmd; return the IntegrationSystem.
ẋ = u ⇒ integrate state by cmd. dim/init size the state (init wins).
unicycle_2d
2-D unicycle dynamics -- ṗ = s·[cos θ, sin θ], θ̇ = ω, ṡ = 0.
UnicycleDynamics
unicycle_2d(world, *, p=None, theta=None, speed=None)
Register unicycle components on world and return [dynamics, integration] systems.
Components: p (N,2) state, theta (N,) state, speed (N,) constant, omega (N,)
command, p_dot (N,2). Integration advances p by p_dot and theta by omega.
utils
Utility functions for ssl_simulator.
add_src_to_path(file=None, relative_path='', deep=0)
Adds the "relative_path" folder to sys.path based on 'file' or actual location.
create_dir(directory)
Create a new directory, and any missing parents, if it doesn't already exist.
An empty path is a no-op, so callers can pass os.path.dirname(filename)
unguarded - a bare filename simply writes to the current directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
directory
|
str
|
The path of the directory to create. |
required |
debug_eig(A, *, include_eigenvectors=True)
Log eigenvalues (and optionally eigenvectors) of a matrix at DEBUG level.
Pure diagnostic -- no-op when DEBUG is disabled.
Parameters
A : np.ndarray Square matrix to decompose. include_eigenvectors : bool, optional If True, also include the eigenvector matrix in the log payload.
get_pprz_idx(data, t, time_label='Time')
Get the index of the first row where the time column is greater than or equal to t.
Parameters
data : pd.DataFrame The DataFrame containing the time series data. t : float The target time value to find in the DataFrame. time_label : str, optional The column name representing time (default is "Time").
Returns
int
The index of the first row where time_label is greater than or equal to t.
Raises
ValueError
If the DataFrame is empty or if no valid index is found.
KeyError
If time_label is not found in the DataFrame.
Example
df = pd.DataFrame({"Time": [0, 1, 2, 3, 4, 5]}) get_idx(df, 2.5) 3
load_class_from_file(module_path, class_name)
Dynamically load a class from a given .py file.
load_pprz_data(filename, t0, tf=None, sep='\t', time_label='Time')
Load data from a Paparazzi .csv file, filtering it based on time range.
Parameters
filename : str The path to the CSV file. t0 : float The start time for the data filter. tf : float, optional The end time for the data filter (default is None, which means no upper time filter). sep : str, optional The delimiter used in the CSV file (default is tab-separated). time_label : str, optional The column name that represents time in the dataset (default is "Time").
Returns
pd.DataFrame A pandas DataFrame containing the filtered data.
parse_kwargs(kwargs_input, kwargs_default)
Merge user-provided keyword arguments with default values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kwargs_input
|
dict
|
Dictionary containing user-specified keyword arguments. |
required |
kwargs_default
|
dict
|
Dictionary containing default keyword arguments. |
required |
Returns
dict: A dictionary where user-specified values override the defaults,
while preserving unspecified default values.
pprz_angle(theta_array)
Convert an angle from standard mathematical coordinates to Paparazzi UAV convention.
Parameters
theta_array : np.ndarray Input angles in radians.
Returns
np.ndarray Converted angles in radians.
Notes
- The Paparazzi UAV convention defines 0 radians as pointing north (upward), whereas standard mathematical convention defines 0 radians as pointing east (rightward).
- This function shifts the angle by -theta + π/2 to align with the Paparazzi convention.
safe_assign(target, source, source_name='dict')
Helper to check for key conflicts during dictionary assignments.
safe_update(target, source, source_name='dict')
Helper to check for key conflicts when updating a dictionary.
validate_dict_attributes(obj, attr_names)
Validate that specified attributes are dictionaries and that callable items have call method.
Parameters
obj : object The object whose attributes should be validated attr_names : list of str Names of attributes to validate
Raises
TypeError If any attribute is not a dict or if callable items lack call method
debug
debug_eig(A, *, include_eigenvectors=True)
Log eigenvalues (and optionally eigenvectors) of a matrix at DEBUG level.
Pure diagnostic -- no-op when DEBUG is disabled.
Parameters
A : np.ndarray Square matrix to decompose. include_eigenvectors : bool, optional If True, also include the eigenvector matrix in the log payload.
dict_ops
parse_kwargs(kwargs_input, kwargs_default)
Merge user-provided keyword arguments with default values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kwargs_input
|
dict
|
Dictionary containing user-specified keyword arguments. |
required |
kwargs_default
|
dict
|
Dictionary containing default keyword arguments. |
required |
Returns
dict: A dictionary where user-specified values override the defaults,
while preserving unspecified default values.
safe_assign(target, source, source_name='dict')
Helper to check for key conflicts during dictionary assignments.
safe_update(target, source, source_name='dict')
Helper to check for key conflicts when updating a dictionary.
validate_dict_attributes(obj, attr_names)
Validate that specified attributes are dictionaries and that callable items have call method.
Parameters
obj : object The object whose attributes should be validated attr_names : list of str Names of attributes to validate
Raises
TypeError If any attribute is not a dict or if callable items lack call method
file_ops
load_class_from_file(module_path, class_name)
Dynamically load a class from a given .py file.
path_ops
add_src_to_path(file=None, relative_path='', deep=0)
Adds the "relative_path" folder to sys.path based on 'file' or actual location.
create_dir(directory)
Create a new directory, and any missing parents, if it doesn't already exist.
An empty path is a no-op, so callers can pass os.path.dirname(filename)
unguarded - a bare filename simply writes to the current directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
directory
|
str
|
The path of the directory to create. |
required |
pprz
get_pprz_idx(data, t, time_label='Time')
Get the index of the first row where the time column is greater than or equal to t.
Parameters
data : pd.DataFrame The DataFrame containing the time series data. t : float The target time value to find in the DataFrame. time_label : str, optional The column name representing time (default is "Time").
Returns
int
The index of the first row where time_label is greater than or equal to t.
Raises
ValueError
If the DataFrame is empty or if no valid index is found.
KeyError
If time_label is not found in the DataFrame.
Example
df = pd.DataFrame({"Time": [0, 1, 2, 3, 4, 5]}) get_idx(df, 2.5) 3
load_pprz_data(filename, t0, tf=None, sep='\t', time_label='Time')
Load data from a Paparazzi .csv file, filtering it based on time range.
Parameters
filename : str The path to the CSV file. t0 : float The start time for the data filter. tf : float, optional The end time for the data filter (default is None, which means no upper time filter). sep : str, optional The delimiter used in the CSV file (default is tab-separated). time_label : str, optional The column name that represents time in the dataset (default is "Time").
Returns
pd.DataFrame A pandas DataFrame containing the filtered data.
pprz_angle(theta_array)
Convert an angle from standard mathematical coordinates to Paparazzi UAV convention.
Parameters
theta_array : np.ndarray Input angles in radians.
Returns
np.ndarray Converted angles in radians.
Notes
- The Paparazzi UAV convention defines 0 radians as pointing north (upward), whereas standard mathematical convention defines 0 radians as pointing east (rightward).
- This function shifts the angle by -theta + π/2 to align with the Paparazzi convention.
processing
Compatibility shim: save/load of the canonical log formats lives in ssl_link now.
The formats themselves (csv + # SETTINGS: header, npz, hdf5) are the data plane's standard
(ssl_link.persistence), shared by simulated and real producers. Import from ssl_link in
new code; this module keeps existing ssl_simulator imports working.