producer-live-data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Producer Script for Live-Data from API-Call"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "#### Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "import json\n",
    "import time\n",
    "from kafka import KafkaProducer, KafkaAdminClient\n",
    "from kafka.admin import NewTopic\n",
    "import requests\n",
    "import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "admin = KafkaAdminClient(bootstrap_servers=\"localhost:9092\")\n",
    "producer = KafkaProducer(bootstrap_servers=\"localhost:9092\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "### Try to create topics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "if \"preise\" not in admin.list_topics() and \"stationen\" not in admin.list_topics():\n",
    "    new_topics= list()\n",
    "    new_topics.append(NewTopic(name=\"preise\", num_partitions=1, replication_factor=1))\n",
    "    new_topics.append(NewTopic(name=\"stationen\", num_partitions=1, replication_factor=1))\n",
    "    admin.create_topics(new_topics)"
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "Functions for creating a json and adding it to the queue"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "def add_price(date, uuid, diesel, e5,e10):\n",
    "    j = json.dumps({\"date\": date, \"station_uuid\":uuid,\"diesel\":diesel,\"e10\":e10,\"e5\":e5})\n",
    "    #print(j)\n",
    "    producer.send(\"preise\",str(j).encode())\n",
    "\n",
    "def add_station( uuid, name, brand, street, house_number, post_code,city, lat, lng):\n",
    "    j = json.dumps({\"uuid\":uuid,\"name\":name,\"brand\":brand,\"street\":street, \"house_number\":house_number,\"post_code\":post_code,\"city\":city,\"latitude\":lat,\"longitude\":lng,\"first_active\":\"\",\"openingtimes_json\":None})\n",
    "    #print(j)\n",
    "    producer.send(\"stationen\",str(j).encode())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "#### API Call"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "API_KEY = \"e82265a8-b634-ed81-69d2-82b1249c39ea\"\n",
    "BASE_URL = \"https://creativecommons.tankerkoenig.de/json/list.php?\"\n",
    "LAT=49.495411\n",
    "LNG=8.461562\n",
    "RAD=10.0\n",
    "\n",
    "def api_call():\n",
    "    response = requests.get(BASE_URL+\"lat=\"+str(LAT)+\"&lng=\"+str(LNG)+\"&rad=\"+str(RAD)+\"&type=all&apikey=\"+API_KEY)\n",
    "    if response.status_code==200 and response.json()[\"ok\"] is True:\n",
    "        response = response.json()\n",
    "        results = response[\"stations\"]\n",
    "        print(\"Got \"+str(len(results))+\" results\")\n",
    "        for r in results:\n",
    "            add_price(date=datetime.datetime.now().strftime(\"%Y-%m-%d %H:%M:%S+01\"), uuid=r[\"id\"], diesel=r[\"diesel\"], e5=r[\"e5\"], e10=r[\"e10\"])\n",
    "            add_station(uuid=r[\"id\"],name=r[\"name\"],brand=r[\"brand\"],street=r[\"street\"], city=r[\"place\"], house_number=r[\"houseNumber\"], post_code=r[\"postCode\"], lat=r[\"lat\"],lng=r[\"lng\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "Make the API Call every minute"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "is_executing": true,
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Got 82 results\n",
      "Got 82 results\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    while 1:\n",
    "        api_call()\n",
    "        time.sleep(60)\n",
    "except KeyboardInterrupt:\n",
    "    print(\"KeyBoardInterrupt\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "Close connection to producer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "producer.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}