Skip to main content
A disturbance anticipator learns the rhythm of a recurring disturbance — something that hits your robot at a roughly regular interval — and tells you, at any moment in time, whether to protect now and which registered protective action is currently best. You feed it the times disturbances occur and the outcomes of protective actions you deploy. It keeps the learned period, phase, and confidence entirely on the server; you only ever see the decision.

​
Opening an anticipator

Register the protective actions it may choose between, and (optionally) tune its timing window. 
from cadenza_cli import MeganTK

tk = MeganTK()

a = tk.anticipator(
    actions=["brace", "dodge"],   # protective actions to choose between
    lead=1.4,                     # seconds to START protecting before a predicted hit
    guard=0.8,                    # seconds to KEEP protecting after
    min_events=2,                 # disturbances to observe before trusting the period
)
print(a.anticipator_id, a.actions)
Like sessions, it’s a context manager: 
with tk.anticipator(actions=["brace", "dodge"]) as a:
    ...
# anticipator deleted automatically

​
Teaching it the rhythm — disturbance

Call disturbance(t) each time the disturbance actually occurs, passing the time (in seconds) it happened. After min_events observations it locks onto the period. 
for t in (0.0, 2.0, 4.0):     # a ~2s periodic disturbance
    status = a.disturbance(t=t)

status   # {'anticipator_id': ..., 'ready': True, 'best_action': 'brace'}
ready flips to True once it has learned enough to act on.

​
Recording what worked — outcome

When you deploy a protective action, tell the anticipator whether it saved the outcome. This is how it learns which action is best; over time best_action reflects what actually protects you. 
a.outcome("brace", saved=True)     # 'brace' preserved the outcome
a.outcome("dodge", saved=False)    # 'dodge' didn't help this time
name must be one of the actions you registered when opening the anticipator. An unregistered name is rejected (HTTP 422 → MeganTKError).

​
The decision — protect

Ask, at a given time t, whether to protect right now: 
p = a.protect(t=5.9)          # just before the next ~6.0s hit

p.should_protect   # True  → engage a protective action now
p.best_action      # 'brace' → which one
p.ready            # True  → the rhythm is learned; if False, treat as advisory
protect() returns a ProtectResult. The predicted time of the next hit, the period, and the confidence are withheld so the rhythm model can’t be reconstructed — you get the actionable answer only.

​
Inspecting and closing

a.status()   # {'anticipator_id', 'ready', 'best_action'}
a.close()    # delete server-side; returns True (idempotent)

​
A full anticipation loop

from cadenza_cli import MeganTK

tk = MeganTK()
with tk.anticipator(actions=["brace", "dodge"], min_events=2) as a:
    for event in control_loop():                  # your real-time loop
        if event.disturbance_detected:
            a.disturbance(t=event.t)

        p = a.protect(t=event.t)
        if p.should_protect and p.ready:
            saved = actuators.engage(p.best_action)   # deploy it
            a.outcome(p.best_action, saved=saved)     # close the learning loop
The anticipator and a token session compose cleanly: run a session to govern task progress while an anticipator guards against a periodic disturbance in the same loop. See Building deep.

​
Next

Usage & metering

Every call is counted — read it back.

Building deep

Control-loop integration, rate limits, and error handling.