Skip to content

local_trigger

local_trigger

LocalTriggerChannel — file/socket trigger for clap and other local events.

Classes

LocalTriggerChannel 

LocalTriggerChannel(
    *,
    bus: Optional[EventBus] = None,
    path: Path | str | None = None,
    poll_s: float = 0.35,
)

Bases: BaseChannel

Watch a JSONL drop file for local events (e.g. double-clap welcome).

Writers append one JSON object per line::

{"event": "welcome_home", "content": "Run welcome-clap sequence"}

The channel polls the file and dispatches :class:ChannelMessage to handlers.

Source code in src/diapason/channels/local_trigger.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    *,
    bus: Optional[EventBus] = None,
    path: Path | str | None = None,
    poll_s: float = 0.35,
) -> None:
    self._bus = bus
    self._path = Path(path) if path else default_trigger_path()
    self._poll_s = poll_s
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._stop = threading.Event()
    self._thread: Optional[threading.Thread] = None
    self._offset = 0
Methods:
emit 
emit(
    content: str,
    *,
    event: str = "welcome_home",
    **extra: Any,
) -> None

Append a trigger event (used by clap listener / CLI).

Source code in src/diapason/channels/local_trigger.py 
103
104
105
106
107
108
def emit(self, content: str, *, event: str = "welcome_home", **extra: Any) -> None:
    """Append a trigger event (used by clap listener / CLI)."""
    self._path.parent.mkdir(parents=True, exist_ok=True)
    payload = {"event": event, "content": content, **extra}
    with self._path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(payload, ensure_ascii=False) + "\n")

Functions: