chapter4_agent.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chapter 4 : Write your first Agent"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The Source class"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The Source is the place where you implement the acquisition of your data\n",
    "\n",
    "Regardless of the type of data, the framework will always consume a Source the same way.<br>\n",
    "Many generic sources are provided by the framework and it's easy to create a new one."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The Source strongly encourages to stream incoming data\n",
    "\n",
    "Whatever the size of the dataset, memory will never be an issue."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's continue with our room sensors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# this source provides us with a CSV line each second\n",
    "from pyngsi.sources.more_sources import SourceSampleOrion\n",
    "\n",
    "# init the source\n",
    "src = SourceSampleOrion()\n",
    "\n",
    "# iterate over the source\n",
    "for row in src:\n",
    "    print(row)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we can see that a row is an instance of a Row class.<p>\n",
    "\n",
    "For the vast majority of the Sources, the provider keeps the same value during the datasource lifetime.<br>\n",
    "We'll go into details in next chapters.<p>\n",
    "\n",
    "**In practice you won't iterate the Source by hand. The framework will do it for you.**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The Agent class"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here comes the power of the framework.<br>\n",
    "By using an Agent you will delegate the processing of the Source to the framework.<p>\n",
    "\n",
    "Basically an Agent needs a Source for input and a Sink for output.<br>\n",
    "It also needs a function in order to convert incoming rows to NGSI entities.<p>\n",
    "\n",
    "Once the Agent is initialized, you can run it !"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's continue with our rooms."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from pyngsi.sources.more_sources import SourceSampleOrion\n",
    "from pyngsi.sink import SinkStdout\n",
    "from pyngsi.agent import NgsiAgent\n",
    "\n",
    "# init the source\n",
    "src = SourceSampleOrion()\n",
    "\n",
    "# for the convenience of the demo, the sink is the standard output\n",
    "sink = SinkStdout()\n",
    "\n",
    "# init the agent\n",
    "agent = NgsiAgent.create_agent(src, sink)\n",
    "\n",
    "#run the agent\n",
    "agent.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here you can see that incoming rows are outputted 'as is'.<br>\n",
    "It's possible because SinkStdout ouputs whatever it receives on its input.<br>\n",
    "But SinkOrion expects valid NGSI entities on its input.<p>\n",
    "\n",
    "So let's define a conversion function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyngsi.sources.source import Row\n",
    "from pyngsi.ngsi import DataModel\n",
    "\n",
    "def build_entity(row: Row) -> DataModel:\n",
    "    id, temperature, pressure = row.record.split(';')\n",
    "    m = DataModel(id=id, type=\"Room\")\n",
    "    m.add(\"dataProvider\", row.provider)\n",
    "    m.add(\"temperature\", float(temperature))\n",
    "    m.add(\"pressure\", int(pressure))\n",
    "    return m"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And use it in the Agent."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# init the Agent with the conversion function\n",
    "agent = NgsiAgent.create_agent(src, sink, process=build_entity)\n",
    "\n",
    "# run the agent\n",
    "agent.run()\n",
    "\n",
    "# obtain the statistics\n",
    "print(agent.stats)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Congratulations ! You have developed your first pyngsi Agent !<p>\n",
    "\n",
    "Feel free to try SinkOrion instead of SinkStdout.<br>\n",
    "Note that you get statistics for free."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Inside your conversion function you can filter input rows just by returning None.<br>\n",
    "For example, if you're not interested in Room3 you could write this function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def build_entity(row: Row) -> DataModel:\n",
    "    id, temperature, pressure = row.record.split(';')\n",
    "    if id == \"Room3\":\n",
    "        return None\n",
    "    m = DataModel(id=id, type=\"Room\")\n",
    "    m.add(\"dataProvider\", row.provider)\n",
    "    m.add(\"temperature\", float(temperature))\n",
    "    m.add(\"pressure\", int(pressure))\n",
    "    return m\n",
    "\n",
    "agent = NgsiAgent.create_agent(src, sink, process=build_entity)\n",
    "agent.run()\n",
    "print(agent.stats)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The side_effect() function\n",
    "\n",
    "As of v1.2.5 the Agent takes the `side_effect()` function as an optional argument.<br>\n",
    "That function will allow to create entities aside of the main flow. Few use cases might need it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def side_effect(row, sink, model) -> int:\n",
    "    # we can use as an input the current row or the model returned by our process() function\n",
    "    m = DataModel(id=f\"Building:MainBuilding:Room:{model['id']}\", type=\"Room\")\n",
    "    sink.write(m.json())\n",
    "    return 1 # number of entities created in the function\n",
    "\n",
    "agent = NgsiAgent.create_agent(src, sink, process=build_entity, side_effect=side_effect)\n",
    "agent.run()\n",
    "print(agent.stats)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This NGSI Agent developed in this chapter is a very basic one.<p>\n",
    "Principles remain the same when things become more complex."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.2 64-bit ('venv': venv)",
   "language": "python",
   "name": "python38264bitvenvvenva41776b7b4c24b1ea249e6dd67506952"
  },
  "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.8.2-final"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}