{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5f74176f",
   "metadata": {},
   "source": [
    "# Time and Sync\n",
    "\n",
    "Robots do not run on one global step: cameras stream, controllers run fast,\n",
    "policies have variable latency. Retriever makes that explicit with two\n",
    "independent per-Flow decisions — a **clock** that decides *when* a Flow wakes,\n",
    "and a **sync policy** on each edge that decides *which* buffered upstream record\n",
    "becomes its input. This notebook builds tiny multi-rate graphs and steps them\n",
    "in-process so you can watch both knobs change what each `step()` sees. No\n",
    "backend, camera, or robot — every cell runs in your process."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a4f2978",
   "metadata": {},
   "source": [
    "> **Running in Colab?** The next cell installs `retriever-core`. From a source\n",
    "> checkout (or once it's already installed) the install is skipped."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b8936b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Colab setup: install retriever-core only if it isn't importable yet.\n",
    "try:\n",
    "    import retriever  # noqa: F401\n",
    "except ImportError:  # pragma: no cover\n",
    "    import subprocess\n",
    "    import sys\n",
    "\n",
    "    subprocess.run(\n",
    "        [sys.executable, \"-m\", \"pip\", \"install\", \"retriever-core\"], check=True\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ef5928c",
   "metadata": {},
   "source": [
    "## Two independent knobs\n",
    "\n",
    "Every Flow answers two separate questions. The clock (`@ Rate/Trigger/Hybrid`)\n",
    "answers *when do I wake?* The sync policy (`sync=` on each edge) answers *which\n",
    "buffered record do I read?* They compose freely, and each one has a plain,\n",
    "inspectable repr."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "928b6a0b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from retriever.flow import Events, Hold, Hybrid, Latest, Rate, Trigger, Window\n",
    "\n",
    "print(\"Clocks — decide WHEN a Flow wakes:\")\n",
    "print(\"  Rate(hz=20)                 ->\", Rate(hz=20), f\"(every {Rate(hz=20).interval:.3f}s)\")\n",
    "print(\"  Trigger('value')            ->\", Trigger(\"value\"))\n",
    "print(\"  Hybrid(hz=5, trigger=[...]) ->\", Hybrid(hz=5, trigger=[\"value\"]))\n",
    "print()\n",
    "print(\"Sync policies — decide WHICH upstream record step() gets:\")\n",
    "print(\"  Latest()                    ->\", Latest())\n",
    "print(\"  Window(20, 0.35s, 'mean')   ->\", Window(buffer_size=20, duration=0.35, agg=\"mean\"))\n",
    "print(\"  Hold(debounce=0.25)         ->\", Hold(debounce=0.25))\n",
    "print(\"  Events(10, 0.35s)           ->\", Events(buffer_size=10, duration=0.35, include_timestamps=False))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a0a291fb",
   "metadata": {},
   "source": [
    "## Clocks decide *when*\n",
    "\n",
    "A `Rate` Flow wakes on its own timer, every tick, whether or not new data\n",
    "arrived. A `Trigger` Flow wakes *only* when its named field arrives. A `Hybrid`\n",
    "Flow wakes on the timer **or** immediately on the event. Here a sensor fires\n",
    "every tick but only *emits* a reading on even ticks; watch which downstream\n",
    "Flows go quiet when nothing new arrives."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4581f4ad",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "from retriever.flow import Flow, Pipeline, io\n",
    "\n",
    "\n",
    "@io\n",
    "class Reading:\n",
    "    value: int\n",
    "\n",
    "\n",
    "class Sensor(Flow[None, Reading]):\n",
    "    \"\"\"Wakes every tick, but only publishes a reading on even ticks.\"\"\"\n",
    "\n",
    "    def reset(self):\n",
    "        self.tick = 0\n",
    "        self.emitted = 0\n",
    "\n",
    "    def step(self, _):\n",
    "        self.tick += 1\n",
    "        if self.tick % 2 == 1:  # emit on the 1st, 3rd, 5th tick...\n",
    "            self.emitted += 1\n",
    "            return Reading(value=self.emitted)\n",
    "        return Reading()  # nothing published this tick\n",
    "\n",
    "\n",
    "class Ticker(Flow[Reading, Reading]):  # @ Rate — wakes every tick\n",
    "    def step(self, r: Reading) -> Reading:\n",
    "        return Reading(value=r.value)\n",
    "\n",
    "\n",
    "class Detector(Flow[Reading, Reading]):  # @ Trigger — wakes only on arrival\n",
    "    def step(self, r: Reading) -> Reading:\n",
    "        return Reading(value=r.value)\n",
    "\n",
    "\n",
    "class RateOrEvent(Flow[Reading, Reading]):  # @ Hybrid — wakes on timer OR event\n",
    "    def step(self, r: Reading) -> Reading:\n",
    "        return Reading(value=r.value)\n",
    "\n",
    "\n",
    "clocks = Pipeline(\"time.clocks\")\n",
    "with clocks:\n",
    "    sensor = Sensor() @ Rate(hz=10)\n",
    "    ticker = Ticker() @ Rate(hz=10)\n",
    "    detector = Detector() @ Trigger(\"value\")\n",
    "    hybrid = RateOrEvent() @ Hybrid(hz=10, trigger=[\"value\"])\n",
    "    clocks.connect(sensor, ticker, sync=Latest())\n",
    "    clocks.connect(sensor, detector, sync=Latest())\n",
    "    clocks.connect(sensor, hybrid, sync=Latest())\n",
    "\n",
    "print(\"tick | Sensor emits | Ticker(Rate) | Detector(Trigger) | RateOrEvent(Hybrid)\")\n",
    "for i in range(6):\n",
    "    res = clocks.step(dt=0.1)\n",
    "    emitted = res.outputs[\"Sensor\"].value\n",
    "    emit_s = f\"v={emitted}\" if emitted is not None else \"  -\"\n",
    "    tick_seen = f\"sees v={res.inputs['Ticker'].value}\"\n",
    "    det = f\"FIRED v={res.inputs['Detector'].value}\" if \"Detector\" in res.executed else \"idle\"\n",
    "    hyb = \"FIRED\" if \"RateOrEvent\" in res.executed else \"idle\"\n",
    "    print(f\"  {i}  |    {emit_s:5}    |  {tick_seen:9} | {det:11} | {hyb}\")\n",
    "clocks.close_stepper()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f73bded4",
   "metadata": {},
   "source": [
    "`Detector` is the only Flow that goes quiet on odd ticks — its `Trigger` clock\n",
    "has nothing to fire on. `Ticker` (`Rate`) and `RateOrEvent` (`Hybrid`) wake\n",
    "every tick regardless. (The in-process stepper wakes `Rate` and `Hybrid` once\n",
    "per tick; the extra thing `Hybrid` buys you — waking *immediately* on an event\n",
    "instead of waiting for the next timer edge — is a live-scheduling property.)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c676f015",
   "metadata": {},
   "source": [
    "## Sync decides *which record*\n",
    "\n",
    "Every edge keeps a timestamped buffer of upstream outputs. The `sync=` policy\n",
    "consumes that buffer at wake time and returns exactly one input. Fan the *same*\n",
    "10 Hz ramp (emitting 1, 2, 3, ...) into three consumers with three policies and\n",
    "compare what each `step()` receives on the same tick."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "14ac4a68",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "@io\n",
    "class Signal:\n",
    "    value: int\n",
    "\n",
    "\n",
    "class Ramp(Flow[None, Signal]):\n",
    "    def reset(self):\n",
    "        self.n = 0\n",
    "\n",
    "    def step(self, _):\n",
    "        self.n += 1\n",
    "        return Signal(value=self.n)\n",
    "\n",
    "\n",
    "class Newest(Flow[Signal, Signal]):\n",
    "    def step(self, s: Signal) -> Signal:\n",
    "        return Signal(value=s.value)\n",
    "\n",
    "\n",
    "class Averaged(Flow[Signal, Signal]):\n",
    "    def step(self, s: Signal) -> Signal:\n",
    "        return Signal(value=s.value)\n",
    "\n",
    "\n",
    "class Recent(Flow[Signal, Signal]):\n",
    "    def step(self, s: Signal) -> Signal:\n",
    "        return Signal(value=s.value)\n",
    "\n",
    "\n",
    "sync = Pipeline(\"time.sync\")\n",
    "with sync:\n",
    "    ramp = Ramp() @ Rate(hz=10)\n",
    "    newest = Newest() @ Rate(hz=10)\n",
    "    averaged = Averaged() @ Rate(hz=10)\n",
    "    recent = Recent() @ Rate(hz=10)\n",
    "    sync.connect(ramp, newest, sync=Latest())\n",
    "    sync.connect(ramp, averaged, sync=Window(buffer_size=20, duration=0.35, agg=\"mean\"))\n",
    "    sync.connect(ramp, recent, sync=Events(buffer_size=10, duration=0.35, include_timestamps=False))\n",
    "\n",
    "print(\"tick | Latest() | Window(mean, 0.35s) | Events(recent 0.35s)\")\n",
    "for i in range(6):\n",
    "    res = sync.step(dt=0.1)\n",
    "    latest_v = res.inputs[\"Newest\"].value\n",
    "    mean_v = res.inputs[\"Averaged\"].value\n",
    "    recent_v = res.inputs[\"Recent\"].value\n",
    "    print(f\"  {i}  |    {latest_v:>2}    |        {mean_v:>4}         | {recent_v}\")\n",
    "sync.close_stepper()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f18c0dc",
   "metadata": {},
   "source": [
    "Same stream, same tick, three different inputs. `Latest()` hands over one fresh\n",
    "value. `Window(agg=\"mean\")` summarizes the last 0.35 s and slides forward.\n",
    "`Events(...)` returns the recent records themselves, for Flows that reason over\n",
    "a short history. The policy is a pure function of the *buffered, timestamped\n",
    "records* — not of any global state."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5ec7d4ed",
   "metadata": {},
   "source": [
    "## Hold: rate-limit a chatty stream\n",
    "\n",
    "`Hold` is a zero-order hold: it repeats the last accepted value. With\n",
    "`debounce=`, it also refuses to accept a new value until that many seconds have\n",
    "passed — a leading-edge rate limiter. Feed the same 10 Hz ramp through\n",
    "`Latest()` and `Hold(debounce=0.25)` side by side."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2539bca5",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Debounced(Flow[Signal, Signal]):\n",
    "    def step(self, s: Signal) -> Signal:\n",
    "        return Signal(value=s.value)\n",
    "\n",
    "\n",
    "hold = Pipeline(\"time.hold\")\n",
    "with hold:\n",
    "    src = Ramp() @ Rate(hz=10)\n",
    "    live = Newest() @ Rate(hz=10)\n",
    "    steady = Debounced() @ Rate(hz=10)\n",
    "    hold.connect(src, live, sync=Latest())\n",
    "    hold.connect(src, steady, sync=Hold(debounce=0.25))\n",
    "\n",
    "print(\"tick | Latest() | Hold(debounce=0.25)\")\n",
    "for i in range(6):\n",
    "    res = hold.step(dt=0.1)\n",
    "    print(f\"  {i}  |    {res.inputs['Newest'].value:>2}    | {res.inputs['Debounced'].value:>2}\")\n",
    "hold.close_stepper()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1f167f2",
   "metadata": {},
   "source": [
    "`Latest()` passes every new reading (1, 2, 3, ...). `Hold(debounce=0.25)`\n",
    "accepts one, then ignores updates for 0.25 s and re-serves the held value —\n",
    "so a downstream Flow sees a calm, rate-limited signal without adding its own\n",
    "state. Clocks and sync are also the determinism boundary: the wall clock\n",
    "decides which records land in a buffer, but every sync policy is a pure\n",
    "function of those records — replay the same trace and every `step()` sees the\n",
    "same input. Next: [Runtime](/concepts/runtime/) connects clocks and sync to\n",
    "validation, in-process stepping, backends, and replay."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}