{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "722dd90f",
   "metadata": {},
   "source": [
    "# Flow fundamentals\n",
    "\n",
    "A `Flow` is the smallest unit of Retriever computation: a typed Python class\n",
    "with local state and one synchronous `step()`. This notebook builds one, runs\n",
    "it as plain Python, then drops it into a graph — with no backend, camera, or\n",
    "robot involved. Every cell runs in-process."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0d072c7",
   "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": "79f0ceda",
   "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": "15cb2d59",
   "metadata": {},
   "source": [
    "## The smallest shape\n",
    "\n",
    "Inputs and outputs are `@io` dataclasses; their fields are the graph ports.\n",
    "`Flow[NumberInput, NumberOutput]` is the type boundary Retriever checks."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "09922e43",
   "metadata": {},
   "outputs": [],
   "source": [
    "from retriever.flow import Flow, io\n",
    "\n",
    "\n",
    "@io\n",
    "class NumberInput:\n",
    "    value: int\n",
    "\n",
    "\n",
    "@io\n",
    "class NumberOutput:\n",
    "    result: int\n",
    "\n",
    "\n",
    "class DoubleFlow(Flow[NumberInput, NumberOutput]):\n",
    "    def step(self, input: NumberInput) -> NumberOutput:\n",
    "        return NumberOutput(result=input.value * 2)\n",
    "\n",
    "\n",
    "flow = DoubleFlow()\n",
    "out = flow.step(NumberInput(value=5))\n",
    "print(out)\n",
    "print(\"ports:\", flow.input_type.__name__, \"->\", flow.output_type.__name__)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c3d0f365",
   "metadata": {},
   "source": [
    "A Flow is callable directly — no pipeline, no clock, no event loop. That makes\n",
    "it unit-testable and breakpoint-friendly on its own."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "41e8a0ee",
   "metadata": {},
   "source": [
    "## `_signals` says what arrived\n",
    "\n",
    "`@io` makes every field `Optional`, so a Flow can receive a *partial* input.\n",
    "`_signals` lists the fields that are actually set, so `step()` can branch on\n",
    "what it got instead of assuming a full observation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "940008d2",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "print(\"empty  _signals:\", NumberInput()._signals)\n",
    "print(\"filled _signals:\", NumberInput(value=7)._signals)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2df47554",
   "metadata": {},
   "source": [
    "## Local state with `reset()`\n",
    "\n",
    "Runtime-local state — counters, buffers, model handles — belongs in `reset()`,\n",
    "which the runtime calls once when a run starts. Keep `__init__` lightweight."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "232cad06",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Counter(Flow[None, NumberOutput]):\n",
    "    def reset(self) -> None:\n",
    "        self.count = 0\n",
    "\n",
    "    def step(self, _) -> NumberOutput:\n",
    "        self.count += 1\n",
    "        return NumberOutput(result=self.count)\n",
    "\n",
    "\n",
    "c = Counter()\n",
    "c.reset()\n",
    "print(\"counter:\", [c.step(None).result for _ in range(3)])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e76d0626",
   "metadata": {},
   "source": [
    "## Put timing in the Pipeline, not the Flow\n",
    "\n",
    "A Flow never decides *when* it runs — the graph does, via a clock (`@`) and a\n",
    "sync policy on each edge. `pipe.step(dt=...)` advances the whole graph one tick\n",
    "at a time, in your process, where you can set a breakpoint inside any `step()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f40e3926",
   "metadata": {},
   "outputs": [],
   "source": [
    "from retriever.flow import Latest, Pipeline, Rate, Trigger\n",
    "\n",
    "\n",
    "class Source(Flow[None, NumberInput]):\n",
    "    def reset(self):\n",
    "        self.n = 0\n",
    "\n",
    "    def step(self, _):\n",
    "        self.n += 1\n",
    "        return NumberInput(value=self.n)\n",
    "\n",
    "\n",
    "pipe = Pipeline(\"flow_fundamentals\")\n",
    "with pipe:\n",
    "    src = Source() @ Rate(hz=10)\n",
    "    dbl = DoubleFlow() @ Trigger(\"value\")\n",
    "    pipe.connect(src, dbl, sync=Latest())\n",
    "\n",
    "for i in range(3):\n",
    "    result = pipe.step(dt=0.1)  # in-process; breakpoint-friendly\n",
    "    print(f\"tick {i}: executed={sorted(result.executed)}\")\n",
    "pipe.close_stepper()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}