01 - kafkatoiguazio.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json,time,os\n",
    "import v3io_frames as v3f\n",
    "from math import sin, cos, sqrt, atan2, radians"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%nuclio cmd -c\n",
    "pip install v3io_frames\n",
    "pip install nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting spec.build.baseImage to 'python:3.6-jessie'\n",
      "%nuclio: setting spec.handler to 'kafkatoiguazio:msg_send'\n"
     ]
    }
   ],
   "source": [
    "%%nuclio config \n",
    "spec.build.baseImage = \"python:3.6-jessie\"\n",
    "spec.handler = \"kafkatoiguazio:msg_send\"\n",
    "##spec.handler ????"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "import nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "%nuclio env -c V3IO_USERNAME=iguazio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "#tablename = os.path.join('/v3io/'+os.getenv('V3IO_HOME')+'/flights/flighttracker')\n",
    "#distance_tbl = os.path.join('/v3io/'+os.getenv('V3IO_HOME')+'/flights/distance_traveled')\n",
    "\n",
    "tablename =  '/'+os.getenv('V3IO_USERNAME')+'/flights/flighttracker'\n",
    "distance_tbl = '/'+os.getenv('V3IO_USERNAME')+'/flights/distance_traveled'\n",
    "                            \n",
    "flight_template =\"call_sign='{call_sign}';heading='{heading}';curr_lat='{lat}';curr_lon='{lon}'; \\\n",
    "            origin_country='{origin_country}';ID='{key}';prev_lat='{prev_lat}';prev_lon='{prev_lon}'\"\n",
    "\n",
    "distance_template = \"ID='{ID}';distance_traveled='{distance}'\"\n",
    "\n",
    "\n",
    "backend = 'kv'\n",
    "\n",
    "\n",
    "# Fin the prvious record\n",
    "def find_previous(ID,context):\n",
    "    kd =   context.client.read(backend, tablename,filter='__name==\"'+ID+'\"')\n",
    "    return kd\n",
    "\n",
    "# Calculate distance traveled since last checkpoint\n",
    "def calculate_distance(ID,curr_lat,curr_lon,prev_lat,prev_lon,context):\n",
    "    #Credit: https://stackoverflow.com/questions/19412462/getting-distance-between-two-points-based-on-latitude-longitude\n",
    "    # approximate radius of earth in km\n",
    "    R = 6373.0\n",
    "\n",
    "    lat1 = float(curr_lat)\n",
    "    lon1 = float(curr_lon)\n",
    "    lat2 = float(prev_lat)\n",
    "    lon2 = float(prev_lon)\n",
    "\n",
    "    dlon = lon2 - lon1\n",
    "    dlat = lat2 - lat1\n",
    "\n",
    "    a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2\n",
    "    c = 2 * atan2(sqrt(a), sqrt(1 - a))\n",
    "\n",
    "    distance = R * c\n",
    "\n",
    "    distance_info={}\n",
    "    distance_info['ID']=ID\n",
    "    distance_info['distance']=distance\n",
    "    flight = distance_template.format(**distance_info)\n",
    "    \n",
    "    try:\n",
    "        context.client.execute(backend,distance_tbl,'update', args={'key': ID , 'expression': flight})\n",
    "    except:\n",
    "        context.logger.debug(\"Failed try again: \" + str(flight))\n",
    "        try:\n",
    "            time.sleep(2)\n",
    "            context.client.execute(backend,distance_tbl,'update', args={'key': ID , 'expression': flight})\n",
    "        except:\n",
    "            context.logger.debug(\"Failed give up \" + str(flight) )\n",
    "    \n",
    "\n",
    "# spec.handler = kafkatoiguazio:msg_send\n",
    "# Does not work?\n",
    "#def msg_send(context,message):\n",
    "    \n",
    "def handler(context,message):\n",
    "    #flight_info=json.loads(message.value)\n",
    "    flight_info=json.loads(message.body)\n",
    "    flight_info['prev_lon']=0\n",
    "    flight_info['prev_lat']=0\n",
    "    ID=flight_info['key']\n",
    "    previous_check=find_previous(ID,context)\n",
    "    if not previous_check.empty:\n",
    "        if previous_check['curr_lon'][0] != 0 and previous_check['curr_lat'][0] and flight_info['lat'] != None and flight_info['lon'] != None:\n",
    "            flight_info['prev_lon']=previous_check['curr_lon'][0]\n",
    "            flight_info['prev_lat']=previous_check['curr_lat'][0]\n",
    "            \n",
    "            try:\n",
    "                calculate_distance(ID,flight_info['lat'],flight_info['lon'],previous_check['curr_lat'][0],previous_check['curr_lon'][0],context)     \n",
    "            except:\n",
    "                context.logger.debug(\"Must have a None in one of the coordinates\")\n",
    "    \n",
    "           \n",
    "    flight = flight_template.format(**flight_info)\n",
    "\n",
    "   \n",
    "    try:\n",
    "        context.client.execute(backend,tablename,'update', args={'key': ID , 'expression': flight})\n",
    "    except:\n",
    "        context.logger.debug(\"Failed try again: \" + str(flight))\n",
    "        try:\n",
    "            time.sleep(2)\n",
    "            context.client.execute(backend,tablename,'update', args={'key': ID , 'expression': flight})\n",
    "        except:\n",
    "            context.logger.debug(\"Failed give up \" + str(flight) )\n",
    "        \n",
    "        \n",
    "    return \"Completed\"\n",
    "\n",
    "\n",
    "\n",
    "def init_context(context):\n",
    "    client = v3f.Client('framesd:8081',\n",
    "                        user='iguazio',\n",
    "                        password='data123',\n",
    "                        container='users')\n",
    "    setattr(context, 'client', client)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[nuclio.deploy] 2019-06-17 11:29:57,449 updating flights\n",
      "[nuclio.deploy] 2019-06-17 11:29:57,489 deploying ...\n",
      "[nuclio.deploy] 2019-06-17 11:29:59,513 (info) Building processor image\n",
      "[nuclio.deploy] 2019-06-17 11:29:59,514 {'imageName': 'nuclio/processor-flights:latest', 'level': 'info', 'message': 'Building processor image', 'name': 'deployer', 'time': 1560770999163.0547}\n",
      "[nuclio.deploy] 2019-06-17 11:30:06,615 (info) Pushing image\n",
      "[nuclio.deploy] 2019-06-17 11:30:06,616 {'from': 'nuclio/processor-flights:latest', 'level': 'info', 'message': 'Pushing image', 'name': 'deployer', 'time': 1560771006433.8271, 'to': 'docker-registry.default-tenant.app.marcelo.iguazio-cd2.com:80/nuclio/processor-flights:latest'}\n",
      "[nuclio.deploy] 2019-06-17 11:30:06,616 (info) Build complete\n",
      "[nuclio.deploy] 2019-06-17 11:30:06,617 {'level': 'info', 'message': 'Build complete', 'name': 'deployer', 'result': {'Image': 'nuclio/processor-flights:latest', 'UpdatedFunctionConfig': {'metadata': {'annotations': {'nuclio.io/generated_by': 'function generated at 17-06-2019 by iguazio from /User/GitHub/nuclio-flights/kafkatoiguazio.ipynb'}, 'labels': {'nuclio.io/project-name': '8c2acc7f-c5da-406d-82ce-cfd5d1eb510b'}, 'name': 'flights', 'namespace': 'default-tenant'}, 'spec': {'build': {'baseImage': 'python:3.6-jessie', 'commands': ['pip install v3io_frames', 'pip install nuclio'], 'functionSourceCode': 'IyBHZW5lcmF0ZWQgYnkgbnVjbGlvLmV4cG9ydC5OdWNsaW9FeHBvcnRlciBvbiAyMDE5LTA2LTE3IDExOjI5CgppbXBvcnQganNvbix0aW1lLG9zCmltcG9ydCB2M2lvX2ZyYW1lcyBhcyB2M2YKZnJvbSBtYXRoIGltcG9ydCBzaW4sIGNvcywgc3FydCwgYXRhbjIsIHJhZGlhbnMKCgp0YWJsZW5hbWUgPSAgJy8nK29zLmdldGVudignVjNJT19VU0VSTkFNRScpKycvZmxpZ2h0cy9mbGlnaHR0cmFja2VyJwpkaXN0YW5jZV90YmwgPSAnLycrb3MuZ2V0ZW52KCdWM0lPX1VTRVJOQU1FJykrJy9mbGlnaHRzL2Rpc3RhbmNlX3RyYXZlbGVkJwogICAgICAgICAgICAgICAgICAgICAgICAgICAgCmZsaWdodF90ZW1wbGF0ZSA9ImNhbGxfc2lnbj0ne2NhbGxfc2lnbn0nO2hlYWRpbmc9J3toZWFkaW5nfSc7Y3Vycl9sYXQ9J3tsYXR9JztjdXJyX2xvbj0ne2xvbn0nOyAgICAgICAgICAgICBvcmlnaW5fY291bnRyeT0ne29yaWdpbl9jb3VudHJ5fSc7SUQ9J3trZXl9JztwcmV2X2xhdD0ne3ByZXZfbGF0fSc7cHJldl9sb249J3twcmV2X2xvbn0nIgoKZGlzdGFuY2VfdGVtcGxhdGUgPSAiSUQ9J3tJRH0nO2Rpc3RhbmNlX3RyYXZlbGVkPSd7ZGlzdGFuY2V9JyIKCgpiYWNrZW5kID0gJ2t2JwoKCmRlZiBmaW5kX3ByZXZpb3VzKElELGNvbnRleHQpOgogICAga2QgPSAgIGNvbnRleHQuY2xpZW50LnJlYWQoYmFja2VuZCwgdGFibGVuYW1lLGZpbHRlcj0nX19uYW1lPT0iJytJRCsnIicpCiAgICByZXR1cm4ga2QKCmRlZiBjYWxjdWxhdGVfZGlzdGFuY2UoSUQsY3Vycl9sYXQsY3Vycl9sb24scHJldl9sYXQscHJldl9sb24sY29udGV4dCk6CiAgICBSID0gNjM3My4wCgogICAgbGF0MSA9IGZsb2F0KGN1cnJfbGF0KQogICAgbG9uMSA9IGZsb2F0KGN1cnJfbG9uKQogICAgbGF0MiA9IGZsb2F0KHByZXZfbGF0KQogICAgbG9uMiA9IGZsb2F0KHByZXZfbG9uKQoKICAgIGRsb24gPSBsb24yIC0gbG9uMQogICAgZGxhdCA9IGxhdDIgLSBsYXQxCgogICAgYSA9IHNpbihkbGF0IC8gMikqKjIgKyBjb3MobGF0MSkgKiBjb3MobGF0MikgKiBzaW4oZGxvbiAvIDIpKioyCiAgICBjID0gMiAqIGF0YW4yKHNxcnQoYSksIHNxcnQoMSAtIGEpKQoKICAgIGRpc3RhbmNlID0gUiAqIGMKCiAgICBkaXN0YW5jZV9pbmZvPXt9CiAgICBkaXN0YW5jZV9pbmZvWydJRCddPUlECiAgICBkaXN0YW5jZV9pbmZvWydkaXN0YW5jZSddPWRpc3RhbmNlCiAgICBmbGlnaHQgPSBkaXN0YW5jZV90ZW1wbGF0ZS5mb3JtYXQoKipkaXN0YW5jZV9pbmZvKQogICAgCiAgICB0cnk6CiAgICAgICAgY29udGV4dC5jbGllbnQuZXhlY3V0ZShiYWNrZW5kLGRpc3RhbmNlX3RibCwndXBkYXRlJywgYXJncz17J2tleSc6IElEICwgJ2V4cHJlc3Npb24nOiBmbGlnaHR9KQogICAgZXhjZXB0OgogICAgICAgIGNvbnRleHQubG9nZ2VyLmRlYnVnKCJGYWlsZWQgdHJ5IGFnYWluOiAiICsgc3RyKGZsaWdodCkpCiAgICAgICAgdHJ5OgogICAgICAgICAgICB0aW1lLnNsZWVwKDIpCiAgICAgICAgICAgIGNvbnRleHQuY2xpZW50LmV4ZWN1dGUoYmFja2VuZCxkaXN0YW5jZV90YmwsJ3VwZGF0ZScsIGFyZ3M9eydrZXknOiBJRCAsICdleHByZXNzaW9uJzogZmxpZ2h0fSkKICAgICAgICBleGNlcHQ6CiAgICAgICAgICAgIGNvbnRleHQubG9nZ2VyLmRlYnVnKCJGYWlsZWQgZ2l2ZSB1cCAiICsgc3RyKGZsaWdodCkgKQogICAgCgogICAgCmRlZiBoYW5kbGVyKGNvbnRleHQsbWVzc2FnZSk6CiAgICBmbGlnaHRfaW5mbz1qc29uLmxvYWRzKG1lc3NhZ2UuYm9keSkKICAgIGZsaWdodF9pbmZvWydwcmV2X2xvbiddPTAKICAgIGZsaWdodF9pbmZvWydwcmV2X2xhdCddPTAKICAgIElEPWZsaWdodF9pbmZvWydrZXknXQogICAgcHJldmlvdXNfY2hlY2s9ZmluZF9wcmV2aW91cyhJRCxjb250ZXh0KQogICAgaWYgbm90IHByZXZpb3VzX2NoZWNrLmVtcHR5OgogICAgICAgIGlmIHByZXZpb3VzX2NoZWNrWydjdXJyX2xvbiddWzBdICE9IDAgYW5kIHByZXZpb3VzX2NoZWNrWydjdXJyX2xhdCddWzBdIGFuZCBmbGlnaHRfaW5mb1snbGF0J10gIT0gTm9uZSBhbmQgZmxpZ2h0X2luZm9bJ2xvbiddICE9IE5vbmU6CiAgICAgICAgICAgIGZsaWdodF9pbmZvWydwcmV2X2xvbiddPXByZXZpb3VzX2NoZWNrWydjdXJyX2xvbiddWzBdCiAgICAgICAgICAgIGZsaWdodF9pbmZvWydwcmV2X2xhdCddPXByZXZpb3VzX2NoZWNrWydjdXJyX2xhdCddWzBdCiAgICAgICAgICAgIAogICAgICAgICAgICB0cnk6CiAgICAgICAgICAgICAgICBjYWxjdWxhdGVfZGlzdGFuY2UoSUQsZmxpZ2h0X2luZm9bJ2xhdCddLGZsaWdodF9pbmZvWydsb24nXSxwcmV2aW91c19jaGVja1snY3Vycl9sYXQnXVswXSxwcmV2aW91c19jaGVja1snY3Vycl9sb24nXVswXSxjb250ZXh0KSAgICAgCiAgICAgICAgICAgIGV4Y2VwdDoKICAgICAgICAgICAgICAgIGNvbnRleHQubG9nZ2VyLmRlYnVnKCJNdXN0IGhhdmUgYSBOb25lIGluIG9uZSBvZiB0aGUgY29vcmRpbmF0ZXMiKQogICAgCiAgICAgICAgICAgCiAgICBmbGlnaHQgPSBmbGlnaHRfdGVtcGxhdGUuZm9ybWF0KCoqZmxpZ2h0X2luZm8pCgogICAKICAgIHRyeToKICAgICAgICBjb250ZXh0LmNsaWVudC5leGVjdXRlKGJhY2tlbmQsdGFibGVuYW1lLCd1cGRhdGUnLCBhcmdzPXsna2V5JzogSUQgLCAnZXhwcmVzc2lvbic6IGZsaWdodH0pCiAgICBleGNlcHQ6CiAgICAgICAgY29udGV4dC5sb2dnZXIuZGVidWcoIkZhaWxlZCB0cnkgYWdhaW46ICIgKyBzdHIoZmxpZ2h0KSkKICAgICAgICB0cnk6CiAgICAgICAgICAgIHRpbWUuc2xlZXAoMikKICAgICAgICAgICAgY29udGV4dC5jbGllbnQuZXhlY3V0ZShiYWNrZW5kLHRhYmxlbmFtZSwndXBkYXRlJywgYXJncz17J2tleSc6IElEICwgJ2V4cHJlc3Npb24nOiBmbGlnaHR9KQogICAgICAgIGV4Y2VwdDoKICAgICAgICAgICAgY29udGV4dC5sb2dnZXIuZGVidWcoIkZhaWxlZCBnaXZlIHVwICIgKyBzdHIoZmxpZ2h0KSApCiAgICAgICAgCiAgICAgICAgCiAgICByZXR1cm4gIkNvbXBsZXRlZCIKCgoKZGVmIGluaXRfY29udGV4dChjb250ZXh0KToKICAgIGNsaWVudCA9IHYzZi5DbGllbnQoJ2ZyYW1lc2Q6ODA4MScsCiAgICAgICAgICAgICAgICAgICAgICAgIHVzZXI9J2lndWF6aW8nLAogICAgICAgICAgICAgICAgICAgICAgICBwYXNzd29yZD0nZGF0YTEyMycsCiAgICAgICAgICAgICAgICAgICAgICAgIGNvbnRhaW5lcj0ndXNlcnMnKQogICAgc2V0YXR0cihjb250ZXh0LCAnY2xpZW50JywgY2xpZW50KQoK', 'noBaseImagesPull': True, 'offline': True, 'registry': 'docker-registry.default-tenant.app.marcelo.iguazio-cd2.com:80'}, 'env': [{'name': 'V3IO_USERNAME', 'value': \"'iguazio'\"}], 'handler': 'kafkatoiguazio:handler', 'platform': {}, 'resources': {}, 'runtime': 'python:3.6'}}}, 'time': 1560771006539.632}\n",
      "[nuclio.deploy] 2019-06-17 11:30:08,635 (info) Function deploy complete\n",
      "[nuclio.deploy] 2019-06-17 11:30:08,636 {'httpPort': 30590, 'level': 'info', 'message': 'Function deploy complete', 'name': 'deployer', 'time': 1560771008550.061}\n",
      "[nuclio.deploy] 2019-06-17 11:30:08,640 done updating flights, function address: 18.196.90.90:30590\n",
      "%nuclio: function deployed\n"
     ]
    }
   ],
   "source": [
    "%nuclio deploy -n flights -p training -v -c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}