feat(ocl): implement an event system for evaluation

Author: tachyonicClock

docs/contributing/faq.md

@@ -32,9 +32,10 @@ A learner should implement the appropriate interface:
 * {py:class}`capymoa.base.AnomalyDetector` for anomaly detectors.
 * {py:class}`capymoa.base.ClassifierSSL` for semi-supervised classifiers.
 * {py:class}`capymoa.base.Classifier` for online continual learning classifiers.
-  Optionally also inheriting from {py:class}`capymoa.ocl.base.TrainTaskAware` or
-  {py:class}`capymoa.ocl.base.TestTaskAware` to support learners that are aware of
-  task identities or task boundaries.
+    Optionally also inheriting from {py:class}`capymoa.base.events.Handler` to
+    subscribe to task-boundary events (for example
+    `capymoa.ocl.evaluation.events.TrainTaskBegin` and
+    `capymoa.ocl.evaluation.events.TestTaskBegin`).
 
 If your method is a wrapper around a MOA learner, you should use the appropriate
 base class:

docs/tutorials.rst

@@ -32,6 +32,7 @@ These tutorials will show you how to get started with the CapyMOA library.
    notebooks/save_and_load_model.ipynb
    notebooks/clustering.ipynb
    notebooks/feature_importance.ipynb
+   notebooks/ocl_event_system.ipynb
 
 Talks
 =====

notebooks/ocl_event_system.ipynb

@@ -0,0 +1,214 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "c337f69e",
+   "metadata": {},
+   "source": [
+    "# Online Continual Learning and Event Handlers\n",
+    "\n",
+    "The OCL model is experimenting with an event-based system to facilitate communication\n",
+    "between different objects. This allows objects to hook into stages of training, evaluation, and inference. A common use is to implement custom metrics or loggers."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "3f7ec1bf",
+   "metadata": {},
+   "source": [
+    "This block shows how to use the event system in isolation:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "id": "c26d92f5",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "<class '__main__.MyEvent'>\n"
+     ]
+    }
+   ],
+   "source": [
+    "from capymoa.base.events import Dispatcher, Event\n",
+    "from dataclasses import dataclass\n",
+    "\n",
+    "\n",
+    "@dataclass\n",
+    "class MyEvent(Event):\n",
+    "    value: int\n",
+    "\n",
+    "\n",
+    "dispatcher = Dispatcher()\n",
+    "\n",
+    "\n",
+    "def print_event_type(event):\n",
+    "    print(type(event))\n",
+    "\n",
+    "\n",
+    "dispatcher.subscribe(MyEvent, print_event_type)\n",
+    "dispatcher.notify(MyEvent(42))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "238f776f",
+   "metadata": {},
+   "source": [
+    "This block shows how to use the event system within the OCL training loop. You can\n",
+    "subscribe to the `None` event to get notified about every event emitted."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "id": "8db5dee1",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "<class 'capymoa.ocl.evaluation.events.TrainBegin'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainTaskBegin'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainBatchPredict'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainBatchPredict'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainBatchPredict'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainBatchPredict'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TestBegin'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TestTaskBegin'>\n",
+      "<class 'capymoa.ocl.evaluation.events.EvalBatchPredict'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TestTaskEnd'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TestEnd'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainTaskEnd'>\n",
+      "<class 'capymoa.ocl.evaluation.events.TrainEnd'>\n"
+     ]
+    }
+   ],
+   "source": [
+    "from capymoa.ocl.evaluation import ocl_train_eval_loop\n",
+    "from capymoa.ocl.datasets import TinySplitMNIST\n",
+    "from capymoa.classifier import NoChange\n",
+    "\n",
+    "dispatcher = Dispatcher()\n",
+    "# Subscribe to all events by using None as the event type\n",
+    "dispatcher.subscribe(None, print_event_type)\n",
+    "\n",
+    "scenario = TinySplitMNIST()\n",
+    "_ = ocl_train_eval_loop(\n",
+    "    NoChange(scenario.schema),\n",
+    "    train_streams=scenario.train_loaders(64)[:1],\n",
+    "    test_streams=scenario.test_loaders(64)[:1],\n",
+    "    dispatcher=dispatcher,\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "a8d1e798",
+   "metadata": {},
+   "source": [
+    "We can use the ``TestTaskBegin`` and ``TrainTaskBegin`` events to implement a\n",
+    "task-incremental learner. To standardising subscribing to a dispatcher our method\n",
+    "implement's `attach_with` from the `Handler` class.\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "id": "10f054c2",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Train task 0 has begun.\n",
+      "Train task 1 has begun.\n",
+      "Train task 2 has begun.\n",
+      "Train task 3 has begun.\n",
+      "Train task 4 has begun.\n",
+      "Accuracy 63.57\n"
+     ]
+    }
+   ],
+   "source": [
+    "from capymoa.base import BatchClassifier\n",
+    "from capymoa.base.events import Dispatcher, Handler\n",
+    "from capymoa.classifier import Finetune\n",
+    "from capymoa.ocl.evaluation.events import TrainTaskBegin, TestTaskBegin\n",
+    "from capymoa.ann import Perceptron\n",
+    "from torch import Tensor\n",
+    "\n",
+    "\n",
+    "class PerceptronTI(BatchClassifier, Handler):\n",
+    "    def __init__(self, schema, n_tasks: int) -> None:\n",
+    "        super().__init__(schema)\n",
+    "        self._classifiers = [Finetune(schema, Perceptron) for _ in range(n_tasks)]\n",
+    "        self._train_task = 0\n",
+    "        self._test_task = 0\n",
+    "\n",
+    "    def batch_train(self, x: Tensor, y: Tensor) -> None:\n",
+    "        self._classifiers[self._train_task].batch_train(x, y)\n",
+    "\n",
+    "    def batch_predict_proba(self, x: Tensor) -> Tensor:\n",
+    "        return self._classifiers[self._test_task].batch_predict_proba(x)\n",
+    "\n",
+    "    def _on_train_task_begin(self, event: TrainTaskBegin) -> None:\n",
+    "        print(f\"Train task {event.train_task} has begun.\")\n",
+    "        self._train_task = event.train_task\n",
+    "\n",
+    "    def _on_test_task_begin(self, event: TestTaskBegin) -> None:\n",
+    "        self._test_task = event.test_task\n",
+    "\n",
+    "    def attach_with(self, dispatcher: Dispatcher) -> Handler:\n",
+    "        dispatcher.subscribe(TrainTaskBegin, self._on_train_task_begin)\n",
+    "        dispatcher.subscribe(TestTaskBegin, self._on_test_task_begin)\n",
+    "        return super().attach_with(dispatcher)\n",
+    "\n",
+    "\n",
+    "learner = PerceptronTI(scenario.schema, n_tasks=5)\n",
+    "results = ocl_train_eval_loop(\n",
+    "    learner,  # If the learner is a Handler, it will be automatically subscribed.\n",
+    "    train_streams=scenario.train_loaders(32),\n",
+    "    test_streams=scenario.test_loaders(32),\n",
+    ")\n",
+    "print(f\"Accuracy {results.accuracy_seen_avg * 100:.2f}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "35fc7fd7",
+   "metadata": {},
+   "source": [
+    "If you use a custom train-test loop with a ``Classifier`` that is also a ``Handler`` you\n",
+    "will need to manually notify the classifier as required."
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "capymoa",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.12.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

src/capymoa/base/__init__.py

@@ -21,6 +21,7 @@
     MOAClassifierSSL,
 )
 from ._batch import Batch
+from . import events
 
 __all__ = [
     "_extract_moa_drift_detector_CLI",
@@ -43,4 +44,5 @@
     "MOAClusterer",
     "MOAPredictionIntervalLearner",
     "PredictionIntervalLearner",
+    "events",
 ]

src/capymoa/base/events.py

@@ -0,0 +1,73 @@
+from typing import Callable, Type, Dict, cast
+from abc import ABC, abstractmethod
+
+
+class Event:
+    """Base class for events emitted by a :py:class:`Dispatcher`."""
+
+
+class Dispatcher:
+    """Publish events to subscribed callbacks.
+
+    Create a source, subscribe one or more handlers for an event type, and
+    dispatch event instances with :py:meth:`notify`.
+
+    >>> source = Dispatcher()
+    >>> def handler(event: Event): print(f"Received event: {event}")
+    >>> source.subscribe(Event, handler)
+    >>> source.notify(Event())
+    Received event: <src.capymoa.base.events.Event...
+
+    Subscriptions are keyed by exact event class.
+    """
+
+    def __init__(self) -> None:
+        """Initialize an empty mapping of event types to callbacks."""
+        self.subscribers: Dict[Type[Event] | None, list[Callable[[Event], None]]] = {}
+
+    def subscribe(
+        self, event_type: Type[Event] | None, callable: Callable[[Event], None]
+    ) -> None:
+        """Register a callback for a specific event class."""
+        callable = cast(Callable[[Event], None], callable)
+        self.subscribers.setdefault(event_type, []).append(callable)
+
+    def unsubscribe(
+        self, event_type: Type[Event] | None, callable: Callable[[Event], None]
+    ) -> None:
+        """Remove a previously registered callback for an event class."""
+        callable = cast(Callable[[Event], None], callable)
+        self.subscribers.get(event_type, []).remove(callable)
+
+    def notify(self, event: Event) -> None:
+        """Notify all subscribers of an event.
+
+        Only subscribers of the exact event type will be notified. Subscribers of parent
+        event types will not be notified.
+        """
+        for subscriber in self.subscribers.get(type(event), []):
+            subscriber(event)
+        for subscriber in self.subscribers.get(None, []):
+            subscriber(event)
+
+
+class Handler(ABC):
+    """Abstract interface for components that consume events.
+
+    A sink encapsulates subscription logic and can attach itself to an
+    :py:class:`Dispatcher`.
+
+    Subclass ``EventSink`` and implement :py:meth:`attach_with` to register
+    the sink's handlers with a source.
+    """
+
+    @abstractmethod
+    def attach_with(self, dispatcher: Dispatcher) -> "Handler":
+        """Attach this sink to an event source.
+
+        Implementations should call
+        :py:meth:`capymoa.base.events.Dispatcher.subscribe` for each
+        event type the sink needs to handle.
+
+        :param dispatcher: The source this sink should subscribe to.
+        """

src/capymoa/ocl/__init__.py

@@ -52,6 +52,6 @@
 Backward Transfer: -0.08
 """
 
-from . import base, datasets, evaluation, util, strategy
+from . import datasets, evaluation, util, strategy
 
-__all__ = ["evaluation", "datasets", "strategy", "base", "util"]
+__all__ = ["evaluation", "datasets", "strategy", "util"]