examples/migration_guide.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Migration Guide"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This jupyter notebook demonstrates how to adapt a [pycycling](https://github.com/zacharyedwardbull/pycycling) device into Bleak-FSM.\n",
    "\n",
    "For this example we will use . Sterzo is a bluetooth steering angle sensor that goes under the front wheel of a bicycle mounted on an indoor trainer. \n",
    "\n",
    "The implementation of Sterzo is at [pycycling/sterzo.py](https://github.com/zacharyedwardbull/pycycling/blob/master/pycycling/sterzo.py) but we only need to look at [examples/sterzo_examples.py](https://github.com/zacharyedwardbull/pycycling/blob/master/examples/sterzo_example.py)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For your convenience, here is `sterzo_example.py`:\n",
    "\n",
    "```python\n",
    "\n",
    "import asyncio\n",
    "from bleak import BleakClient\n",
    "\n",
    "from pycycling.sterzo import Sterzo\n",
    "\n",
    "\n",
    "async def run(address):\n",
    "    async with BleakClient(address) as client:\n",
    "        def steering_handler(steering_angle):\n",
    "            print(steering_angle)\n",
    "\n",
    "        await client.is_connected()\n",
    "        sterzo = Sterzo(client)\n",
    "        sterzo.set_steering_measurement_callback(steering_handler)\n",
    "        await sterzo.enable_steering_measurement_notifications()\n",
    "        await asyncio.sleep(60)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    import os\n",
    "\n",
    "    os.environ[\"PYTHONASYNCIODEBUG\"] = str(1)\n",
    "\n",
    "    device_address = \"36A444C9-2A18-4B6B-B671-E0A8D3DADB1D\"\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(run(device_address))\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To begin with, we'll import the same modules plus Bleak-FSM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "from pycycling.sterzo import Sterzo\n",
    "\n",
    "from bleak_fsm import BleakModel\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = BleakModel() # This represents a given bluetooth adapter. Let's assume there's one on the system."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By default, `model.wrap` is a function that does nothing and just returns the input.\n",
    "That's fine if we want to use the `client`, which is an instance of `BleakClient` directly.\n",
    "\n",
    "However, we want to take advantage of the work that went into creating the `Sterzo` implementation.\n",
    "Therefore we need to modify the `model.wrap` function as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.wrap = lambda client: Sterzo(client) # Wrap the model with the Sterzo class"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you look at [line 54 of sterzo.py](https://github.com/zacharyedwardbull/pycycling/blob/a5998eb191edb1b2f2591d26b06f442a4f1ccff3/pycycling/sterzo.py#L54), you can see that the steering measurement callback is called with one positional argument, called `steering_value`. \n",
    "The steering measurement callback is set according through `set_steering_measurement_callback()` ([line 46-47](https://github.com/zacharyedwardbull/pycycling/blob/a5998eb191edb1b2f2591d26b06f442a4f1ccff3/pycycling/sterzo.py#L46-L47))\n",
    "\n",
    "We are responsible for creating and passing in a callback function that will become `self._steering_measurement_callback`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handle_sterzo_measurement(value):\n",
    "    print(f\"Sterzo angle: {value}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.set_measurement_handler = lambda client: client.set_steering_measurement_callback(handle_sterzo_measurement)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we set the `enable_notifications` and `disable_notifications` functions to their corresponding functions in [examples/sterzo_examples.py](https://github.com/zacharyedwardbull/pycycling/blob/master/examples/sterzo_example.py)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.enable_notifications = lambda client: client.enable_steering_measurement_notifications()\n",
    "model.disable_notifications = lambda client: client.disable_steering_measurement_notifications()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using our new Bleak-FSM interface to Sterzo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await BleakModel.start_scan()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await asyncio.sleep(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await BleakModel.stop_scan()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(BleakModel.bt_devices)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since Elite Sterzo is not registered on any official Bluetooth spec, the service UUID is chosen arbitrary by them. This service UUID is found by Googling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "STERZO_SERVICE_UUID = '347b0001-7635-408b-8918-8ff3949ce592'\n",
    "\n",
    "sterzo_address = None\n",
    "# Look for Sterzo device based on whether it advertises the Sterzo service\n",
    "for address, (ble_device, advertisement_data) in BleakModel.bt_devices.items():\n",
    "    if STERZO_SERVICE_UUID in advertisement_data.service_uuids:\n",
    "        sterzo_address = address\n",
    "        print(f\"Found Sterzo at {sterzo_address}\")\n",
    "        break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if sterzo_address is not None:\n",
    "    await model.set_target(sterzo_address)\n",
    "else:\n",
    "    raise Exception(\"No sterzo found. Make sure it's on and the system bluetooth is on.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "success = await model.connect()\n",
    "if not success:\n",
    "    raise Exception(\"Could not connect to Sterzo\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await model.stream()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await asyncio.sleep(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "await model.clean_up()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}