ExoStack / Extend

ExoCore plug-ins

Every algorithm in ExoCore is a plug-in: a Python file holding one class, placed in a folder of ~/exo_stack_resources and named in a YAML file. The node imports it by that name when it starts. Nothing is installed or rebuilt, and PASE lists the new file in its drop-down the next time it starts.

How to read the templatesIn every template, MYSENS1 marks a value you change; the number points to its explanation below the template.

Plug-in	Folder (in ~/exo_stack_resources)	Class name	Constructor	Method called	Named in
Gait-event detector	`algo/gait_events/`	= file name	`(window_size, parameters)`	`detect(values) → 1 \| 0`	`gait_events.yaml → algorithm`
Profile builder	`algo/builder/`	= file name, ends in `_List` or `_Function`	`()`	`buildprofile(values) → list`	`profile.yaml → profile_builder_algo`
Lookup algorithm	`algo/lookup_algo/`	= file name	`(json_path)`	`readlookuptable(args) → list`	`profile.yaml → lookup_algo`
HIL optimizer	`algo/hil_optimizer/`	= file name, UPPER CASE	`(**hil_hyperparams)`	`get_next_candidate()`, `submit_result()`, `get_best_so_far()`, `is_done()`	`profile.yaml → hil_algo`
Cost function	`algo/cost/`	= file name	`(*cost_params)`	`compute(data) → float`	`profile.yaml → cost_algo`
Planner	`planner/`	= file name, ends in `List` or `Function`	`()`	`next_point(t) → float`; reads `self.profile`	`mplanner.yaml → planner_algorithm`

THE THREE NAMING RULES

The class name is identical to the file name without .py.
Builders end in _List or _Function, planners in List or Function; a builder and the planner it feeds must end in the same word (and match profile_type).
Optimizer files and classes are written in upper case, the node converts the configured name to upper case.

A. Gait-event detector

algo/gait_events/RisingEdge.py

class RisingEdge¹:

    def __init__(self, window_size=10, parameters=None):
        self.window_size = window_size                      # history_size in gait_events.yaml
        p = parameters or []                                # parameters list in gait_events.yaml
        self.threshold² = p[0] if len(p) > 0 else 0.5³
        self.window = []

    def detect(self, values):
        # values: one float per [hub, channel] listed under 'sensors', same order
        self.window.append(values[0])
        if len(self.window) > self.window_size:
            self.window.pop(0)
        if len(self.window) < 2:
            return 0
        fired = self.window[-2] <= self.threshold < self.window[-1⁴]
        return 1 if fired else 0                            # 1 = publish one event pulse

1	class name = file name; write it under `algorithm`
2	your parameter names …
3	… and their defaults, read from `parameters` in order
4	the detection rule, evaluated once per synchronized sample

B. Profile builder

algo/builder/Ramp_List.py

class Ramp¹_List:

    def __init__(self):
        self.default_value = 20.0                          # level at the start and end of the profile

    def buildprofile(self, values):
        # values: list from the lookup table (Deployment) or the optimizer (HIL)
        if len(values) != 4²:
            raise ValueError("expected [start, rise_end, peak, end]")
        start, rise_end, peak, end = values³
        points = [(start, self.default_value), (rise_end, peak), (end, self.default_value)⁴]
        profile = [1.0, start]                              # 1.0 = List profile, then the start time
        for t, v in points:
            profile += [t, v]                               # waypoints, times strictly increasing
        profile.append(end)                                 # end time
        return profile

1	name ending in `_List` (waypoints) or `_Function` (coefficients)
2	number of input values: the optimizer dimension or the length of each lookup-table row
3	the meaning of each input value
4	the waypoints (time or gait phase, value). A `_Function` builder returns `[2.0, start, c0, c1, …, cn, end]` instead

C. Lookup algorithm

algo/lookup_algo/Lookup_Speed_Nearest.py

import json

class Lookup_Speed_Nearest¹:

    def __init__(self, json_path):                         # file named by lookup_table in profile.yaml
        with open(json_path) as f:
            self.table = json.load(f)
        self.keys = sorted(float(k) for k in self.table)

    def readlookuptable(self, args):
        # args: [speed], or [speed, intent] when intent_required is true
        x = float(args[0])
        key = min(self.keys, key=lambda k: abs(k - x))²
        return self.table[str(key)]                         # list of builder input values

1	class name = file name
2	the selection rule: nearest key here; interpolate between keys, or index by intent first

D. HIL optimizer

algo/hil_optimizer/OPT_RANDOM.py

import random

class OPT_RANDOM¹:

    def __init__(self, bounds=None, max_iters=20, maximize=True²):  # = keys of hil_hyperparams
        self.bounds, self.max_iters, self.maximize = bounds, max_iters, maximize
        self.n = 0
        self.best_point, self.best_score = None, None

    def get_next_candidate(self):
        # one value per builder input
        return [lo + (hi - lo) * random.random() for lo, hi in self.bounds³]

    def submit_result(self, candidate, cost):
        # averaged cost of the candidate just evaluated
        self.n += 1
        if self.best_score is None or (cost > self.best_score) == self.maximize:
            self.best_point, self.best_score = candidate, cost
        pass⁴                                          # update your model here

    def get_best_so_far(self):
        return self.best_point, self.best_score

    def is_done(self):                                      # True: results are saved and the node stops
        return self.n >= self.max_iters

1	upper-case name = file name
2	constructor arguments; each key under `hil_hyperparams` must appear here
3	how the next candidate is proposed
4	how a new (candidate, cost) pair updates your optimizer

The node waits for the cost of each candidate before asking for the next one, and discards any cost carrying a different candidate number, so submit_result always receives the cost of the candidate it proposed.

E. Cost function

algo/cost/PeakForceTracking.py

import numpy as np

class PeakForceTracking¹:

    def __init__(self, target_force²):                   # cost_params in profile.yaml, in order
        self.target = target_force

    def compute(self, data):
        # data: NumPy array, one row per sample of ONE gait cycle,
        #       one column per [hub, channel] under 'sensors' in profile.yaml
        force = data[:, 2³]
        return -abs(float(np.max(force)) - self.target)⁴

1	class name = file name
2	constructor arguments, filled positionally from `cost_params`
3	column of the signal you need (order of `sensors`; a gait-marker column is appended last if configured)
4	the per-cycle cost; the node averages it over `gait_cycles_to_consider` cycles

F. Planner

planner/NearestList.py

class Nearest¹List:

    def __init__(self):
        self.profile = None
        # set by the planner node before every call:
        #   List planners:     [[t1, v1], [t2, v2], ...]
        #   Function planners: [c0, c1, ..., cn]

    def next_point(self, t):
        # t: ms since the gait event (time_based) or gait phase (gait_phase_based)
        pts = self.profile
        if t <= pts[0][0]:
            return pts[0][1]
        if t >= pts[-1][0]:
            return pts[-1][1]
        for (t0, v0), (t1, v1) in zip(pts, pts[1:]):
            if t0 <= t <= t1:
                return v0 if (t - t0) < (t1 - t) else v1²

1	name ending in `List` or `Function`
2	the interpolation rule: nearest waypoint here; linear, PCHIP, or a polynomial for Function planners

TRY A PLUG-IN WITHOUT THE ROBOTEvery plug-in is plain Python and can be imported and called in an interpreter before it is used in the stack, e.g. from Ramp_List import Ramp_List; Ramp_List().buildprofile([0, 150, 60, 400]). After saving the file, restart the stage that uses it (restart icon in the Launch tab).