Edge Workers

cyberwave edge install --without-workers

cyberwave worker list                   # list installed workers (shows origin)
cyberwave worker add detect_people.py  # copy a file into the workers dir
cyberwave worker remove detect_people  # remove a worker (with or without .py)
cyberwave worker status                # show worker files + container state
cyberwave worker logs                  # stream worker container logs
cyberwave worker monitor               # live resource/throughput dashboard
cyberwave worker doctor                # diagnose silent failure modes

Installed Workers (~/.cyberwave/workers)
┌───────────────────────────┬──────────┬────────────────────────────┬───────┐
│ Name                      │ Origin   │ File                       │ Size  │
├───────────────────────────┼──────────┼────────────────────────────┼───────┤
│ detect_people             │ custom   │ detect_people.py           │ 512 B │
│ wf_a1b2c3d4               │ workflow │ wf_a1b2c3d4.py             │ 1.2 kB│
└───────────────────────────┴──────────┴────────────────────────────┴───────┘

cyberwave-edge-core worker restart

import logging
logging.getLogger("cyberwave.workflows").setLevel(logging.WARNING)

cyberwave worker monitor               # default 2s refresh
cyberwave worker monitor --update 1    # 1s refresh
cyberwave worker monitor -c <name>     # explicit container

cyberwave workflow list                # list workflows with status and target twin(s)
cyberwave workflow list --json         # JSON output
cyberwave workflow show                # interactive selection, shows nodes and twins
cyberwave workflow show <uuid>         # show specific workflow
cyberwave workflow create -n "Name"    # create a workflow
cyberwave workflow create --template motion-detection
cyberwave workflow activate            # activate (interactive selector)
cyberwave workflow deactivate          # deactivate (interactive selector)
cyberwave workflow sync                # sync workflow to edge device(s) via MQTT
cyberwave workflow sync --edge-active  # restrict selector to active edge workflows
cyberwave workflow sync <uuid>         # sync specific workflow
cyberwave workflow delete --yes        # delete (skip confirmation)

camera_frame → call_model → detection_event_gate → timed_condition → send_alert

# 1. Copy the generated worker to a new custom name
cp ~/.cyberwave/workers/wf_a1b2c3d4.py \
   ~/.cyberwave/workers/my_detector.py

# 2. Edit the new custom worker
nano ~/.cyberwave/workers/my_detector.py

# 3. Deactivate the originating workflow in the UI
#    (this removes wf_a1b2c3d4.py on the next sync)

# ~/.cyberwave/workers/detect_people.py
# ``cw`` is injected by the worker runtime — no import needed.
# For IDE support, uncomment: from cyberwave import Cyberwave; cw: Cyberwave

model = cw.models.load("yolov8n")        # cached, safe to call at module level
twin_uuid = cw.config.twin_uuid

@cw.on_frame(twin_uuid, sensor="front")
def on_frame(frame, ctx):
    results = model.predict(frame, classes=["person"], confidence=0.5)
    for det in results:
        cw.publish_event(twin_uuid, "person_detected", {
            "confidence": getattr(det, "score", None),
            "frame_ts": ctx.timestamp,
        })

"""
Generated edge worker for workflow: Alert on Person
Workflow UUID: a1b2c3d4-...

Auto-generated by WorkerCodegen — DO NOT EDIT.
To customise this worker, eject it:
  cp wf_a1b2c3d4.py alert_on_person.py
"""

import time

# ``cw`` is injected by the worker runtime — no import needed.
# For IDE support, uncomment: from cyberwave import Cyberwave; cw: Cyberwave

yolov8n = cw.models.load("yolov8n.pt")  # type: ignore[name-defined]  # noqa: F821

_last_emit_ts_0 = 0.0
_prev_labels_0 = set()

@cw.on_frame("twin-uuid", sensor="front", fps=5)  # type: ignore[name-defined]  # noqa: F821
def on_frame_a1b2c3d4_0(frame, ctx):
    """Camera frame handler for workflow 'Alert on Person'."""
    global _last_emit_ts_0, _prev_labels_0
    results = yolov8n.predict(frame, classes=["person"], confidence=0.5)
    current_labels = {
        getattr(d, 'label', getattr(d, 'class_name', ''))
        for d in results
    }
    new_labels = current_labels - _prev_labels_0
    if not new_labels:
        _prev_labels_0 = current_labels
        return
    now = time.monotonic()
    if now - _last_emit_ts_0 < 5.0:
        return
    _prev_labels_0 = current_labels
    _last_emit_ts_0 = now
    for det in results:
        label = getattr(det, 'label', getattr(det, 'class_name', ''))
        if label not in new_labels:
            continue
        cw.publish_event(  # type: ignore[name-defined]  # noqa: F821
            "twin-uuid",
            "person_detected",
            {
                "severity": "WARNING",
                "model": "yolov8n.pt",
                "confidence": getattr(det, 'score', None),
                "frame_ts": ctx.timestamp,
            },
        )

import time, zenoh

cfg = zenoh.Config()
cfg.insert_json5("connect/endpoints", '["tcp/<edge-host>:7447"]')
session = zenoh.open(cfg)

seen = {}
def on(sample):
    seen[str(sample.key_expr)] = seen.get(str(sample.key_expr), 0) + 1

sub = session.declare_subscriber("**", on)
time.sleep(3)
for k, n in sorted(seen.items(), key=lambda kv: -kv[1]):
    print(f"{n:>6}  {k}")