TD4/TD_4.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "TD_4.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ocvIoR-mFya7",
        "colab_type": "text"
      },
      "source": [
        "# **Twitter TD 4**\n",
        "\n",
        "Dates: 2019\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "O3j0nwGjGZBo",
        "colab_type": "text"
      },
      "source": [
        "## Initial Setup\n",
        "\n",
        "- **Run \"Setup\" below first.**\n",
        "\n",
        "    - This will load libraries and download some resources that we'll use throughout the tutorial.\n",
        "\n",
        "    - You will see a message reading \"Done with setup!\" when this process completes.\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GKVEnPi34qj4",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "#@title Setup (click the \"run\" button to the left) {display-mode: \"form\"}\n",
        "\n",
        "## Setup ##\n",
        "\n",
        "# imports\n",
        "\n",
        "# built-in Python libraries\n",
        "# -------------------------\n",
        "\n",
        "! pip install peony-twitter[all]\n",
        "import time\n",
        "import json\n",
        "import datetime\n",
        "from peony import EventStream, PeonyClient, event_handler, events\n",
        "import asyncio\n",
        "import random\n",
        "import sys\n",
        "import csv\n",
        "\n",
        "print(\"Done with setup!\")\n",
        "print(\"If you'd like, you can click the (X) button to the left to clear this output.\")\n",
        "\n",
        "# generate graph from twitter json\n",
        "def create_graph(data):\n",
        "    all_data = dict(data)\n",
        "    edge_list = []\n",
        "    user_screen_name = all_data['user']['screen_name']\n",
        "    timestamp_ms = all_data['timestamp_ms']\n",
        "    if len(all_data['entities']['hashtags']) > 0:\n",
        "        for index, i in enumerate(all_data['entities']['hashtags']):\n",
        "            edge_list.append((\"@\"+user_screen_name, \"#\"+all_data['entities']['hashtags'][index]['text'], timestamp_ms))\n",
        "    if len(all_data['entities']['user_mentions']) > 0:\n",
        "        for index, i in enumerate(all_data['entities']['user_mentions']):\n",
        "            edge_list.append((\"@\"+user_screen_name, \"@\"+all_data['entities']['user_mentions'][index]['screen_name'], timestamp_ms))\n",
        "    return edge_list\n",
        "  \n",
        "# compute components from reservoir of edges\n",
        "def components(edges):\n",
        "  global d\n",
        "  d={}\n",
        "  n=0\n",
        "  for e in edges:\n",
        "     a =e [0]\n",
        "     b = e[1]\n",
        "     if a in d.keys():\n",
        "        d[a].add(b)\n",
        "     else:\n",
        "      d[a]=set([b])\n",
        "     if b in d.keys():\n",
        "      d[b].add(a)\n",
        "     else:\n",
        "      d[b]=set([a])\n",
        "\n",
        "  #print (\"Dict=\", d, len(d))\n",
        "  n=len(d)\n",
        "  m=0\n",
        "  #Breadth-first search from first point, first component\n",
        "  a=list(d.keys())[0]\n",
        "\n",
        "  #dc keeps b:i  if i is the length of the shortest path from a to b in the first component\n",
        "  dc={}\n",
        "  dc[a]=0\n",
        "  for b in d[a]: \n",
        "    if b in dc.keys(): pass\n",
        "    else: dc[b]=1\n",
        "\n",
        "  #Initialize S ans S1, Start iterating\n",
        "  S=d[a]\n",
        "  comp=[]\n",
        "  S1=set([a])\n",
        "  S=S.union(S1)\n",
        "  #print(\"S=\",S)\n",
        "  while S > S1:\n",
        "      S1=S\n",
        "      for u in S:\n",
        "         S=S.union(d[u])\n",
        "         for v in d[u]:\n",
        "           if v in dc.keys():\n",
        "              dc[v]=min(dc[v],dc[u]+1)\n",
        "           else:\n",
        "            dc[v]=dc[u]+1\n",
        "  for u in S:\n",
        "       m=m+len(d[u])\n",
        "\n",
        "  comp.append((len(S),int(m/2),list(S)))\n",
        "    \n",
        "  #print(\"Component\",comp)\n",
        "\n",
        "  ST=S\n",
        "    \n",
        "  #print(\"ST=\",ST)\n",
        "\n",
        "  #The other components: origin must be outside ST, same treatment\n",
        "  i=1\n",
        "  while i<len(d):\n",
        "   m=0\n",
        "   while  list(d.keys())[i] not in ST:\n",
        "    a=list(d.keys())[i]\n",
        "    S=d[a]\n",
        "    S1=set([a])\n",
        "    S=S.union(S1)\n",
        "    while S > S1:\n",
        "      S1=S\n",
        "      for u in S:\n",
        "         S=S.union(d[u])\n",
        "    for u in S:\n",
        "       m=m+len(d[u])\n",
        "    comp.append((len(S),int(m/2),list(S)))\n",
        "    ST=ST.union(S)  \n",
        "   i+=1     \n",
        "  return comp\n",
        "\n",
        "\n",
        "def comp_edges(comp1):\n",
        " compedges=[]\n",
        " j0=0\n",
        " while j0<len(comp1):\n",
        "   l1 = comp1[j0][2]\n",
        "   j=0\n",
        "   cp=[]\n",
        "   while j <len(l1):\n",
        "     a=l1[j]\n",
        "     j1=0\n",
        "     while j1< len(d[l1[j]]):\n",
        "      f=(a,list(d[l1[j]])[j1])\n",
        "      cp.append(f)\n",
        "      j1 +=1\n",
        "     j +=1\n",
        "   j0 +=1\n",
        "   compedges.append(cp)\n",
        " return compedges\n",
        "\n",
        "# write edges into file\n",
        "\n"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HBFNVbQ3NBEj",
        "colab_type": "text"
      },
      "source": [
        "## The Twitter API "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-FFngrFrM7sN",
        "colab_type": "text"
      },
      "source": [
        "- Twitter is also known for being an abundant source of publc text data (perhaps even more so than Reddit).\n",
        "- Twitter provides several types of API that can be used to collect anything from tweets to user descriptions to follower networks.\n",
        "    - You can [read all about it here](https://developer.twitter.com/).\n",
        "- For this tutorial, we'll look at using the [streaming API](https://developer.twitter.com/en/docs/tutorials/consuming-streaming-data.html), which allows us to retreive tweets that contain specific words, phrases, and hashtags.\n",
        "- In the slides, we talked about how to setup a Twitter App and get a API keys.\n",
        "    - You should add your own keys below and then run the code block to set your keys:\n",
        "    \n",
        "#### [Tutorial link to create an Twitter app](https://botwiki.org/resource/tutorial/how-to-create-a-twitter-app/) or https://www.youtube.com/watch?v=aB1x7ZsbbaQ&feature=youtu.be"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uVXtia_JJJNu",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "\n",
        "# Go to http://dev.twitter.com and create an app. \n",
        "# The consumer key and secret will be generated for you after\n",
        "consumer_key=\"\"\n",
        "consumer_secret=\"\"\n",
        "\n",
        "# After the step above, you will be redirected to your app's page.\n",
        "# Create an access token under the the \"Your access token\" section\n",
        "access_token=\"\"\n",
        "access_token_secret=\"\"\n"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2Ui2fEzDNbKc",
        "colab_type": "text"
      },
      "source": [
        "- Do not share your credentials with anyone!\n",
        "    - You shouldn't hardcode your API keys in code (like above) if you are going to save the file anywhere that is visible to others (like commiting the file to github).\n",
        "        - You can read more about securing your API keys [here](https://developer.twitter.com/en/docs/basics/authentication/guides/securing-keys-and-tokens).\n",
        "     - So, if you plan to save this file in any way, make sure to remove your API keys first.\n",
        "     - If you think your keys have been compromized, you can regenerate them.\n",
        "        - [Apps](https://developer.twitter.com/en/apps) -> Keys and Tokens -> Regenerate"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XuWUEnMkVm1Z",
        "colab_type": "text"
      },
      "source": [
        "## Capture Data"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rYNTJn478ue3",
        "colab_type": "text"
      },
      "source": [
        "![Step reservoir sampling](https://github.com/GuillaumeVIMONT/TD_UP2_new/blob/master/static/img/continuous_sampling.png?raw=true)\n",
        "\n",
        "Before start Twitter capture we need some informations :\n",
        "\n",
        "* $k$ is the size of the reservoir \\\\\n",
        "* $tau$ is the lenght of the window \\\\\n",
        "* $lambda$ is the lenght of a strate \\\\\n",
        "* $tracking$ is the keywords to filter Twitter stream \\\\\n",
        "* $threshold$ is the threshold for the connected component export \\\\\n",
        "* $timeout$ is the duration of the capture"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2jy0CtGEIdCI",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "#@title Capture Twitter streaming (double-click to view) {display-mode: \"form\"}\n",
        "\n",
        "k = 400 #@param {type:\"number\"}\n",
        "tau = 3 #@param {type:\"number\"}\n",
        "lamb = 1 #@param {type:\"number\"}\n",
        "tracking = \"bitcoin\" #@param {type:\"string\"}\n",
        "threshold = 15 #@param {type:\"slider\", min:1, max:20, step:1}\n",
        "timeout = 15 #@param {type:\"slider\", min:1, max:120, step:1}\n",
        "timeout = int(timeout)*60000\n",
        "start = int(round(time.time() * 1000))\n",
        "\n",
        "global L, sample_window_stream, global_counter, time_counter, interval_counter\n",
        "L=0\n",
        "sample_window_stream = []\n",
        "global_counter = 0\n",
        "interval_counter = 0\n",
        "time_counter = (time.time()*1000)+60000\n",
        "\n",
        "\n",
        "# k is the size of the reservoir (number of edges into the reservoir)\n",
        "\n",
        "\n",
        "# tau is the lenght of the window\n",
        "tau =  tau*60000\n",
        "\n",
        "# lamb is the lenght of strate floato the window\n",
        "lamb = lamb*60000\n",
        "\n",
        "# rate is the number of strates per window\n",
        "rate = tau/lamb\n",
        "\n",
        "# M contain steps floato the window, for each strates we have a edges counter\n",
        "M = [0] * int(rate)\n",
        "\n",
        "# init it's a default parameters to configure somes variables when the reservoir strart\n",
        "init = 0\n",
        "\n",
        "reservoir = []\n",
        "\n",
        "def diameter(i):\n",
        " comp = components(reservoir)\n",
        " comp1 = sorted(comp, reverse=True)\n",
        " #print(\"number of vertices in largest component\",comp1[0][0])\n",
        " #print(\"number of edges in largest component\",comp1[0][1])\n",
        " first=comp1[i]\n",
        " l=first[2]\n",
        " a=l[0]\n",
        "#print(\"First point\",a)\n",
        " dc={}\n",
        " dc[a]=0\n",
        " for b in d[a]: \n",
        "  if b in dc.keys(): pass\n",
        "  else: dc[b]=1\n",
        "  \n",
        "#Initialize S ans S1, Start iterating\n",
        " S=d[a]\n",
        " comp=[]\n",
        " S1=set([a])\n",
        " S=S.union(S1)\n",
        "#print(\"S=\",S)\n",
        " while S > S1:\n",
        "    S1=S\n",
        "    for u in S:\n",
        "       S=S.union(d[u])\n",
        "       for v in d[u]:\n",
        "         if v in dc.keys(): dc[v]=min(dc[v],dc[u]+1)\n",
        "         else: dc[v]=dc[u]+1\n",
        " dist=list(dc.values())\n",
        " ma=max(dist)\n",
        " for a, i in dc.items():\n",
        "   if i==ma: last=a\n",
        "   \n",
        "# we keep last  !!  \n",
        "# do it again !!\n",
        "\n",
        " a=last\n",
        " dc={}\n",
        " dc[a]=0\n",
        " for b in d[a]: \n",
        "  if b in dc.keys(): pass\n",
        "  else: dc[b]=1\n",
        "  \n",
        "#Initialize S ans S1, Start iterating\n",
        " S=d[a]\n",
        " comp=[]\n",
        " S1=set([a])\n",
        " S=S.union(S1)\n",
        "#print(\"S=\",S)\n",
        " while S > S1:\n",
        "    S1=S\n",
        "    for u in S:\n",
        "       S=S.union(d[u])\n",
        "       for v in d[u]:\n",
        "         if v in dc.keys(): dc[v]=min(dc[v],dc[u]+1)\n",
        "         else: dc[v]=dc[u]+1\n",
        " dist=list(dc.values())\n",
        " ma=max(dist)\n",
        " for a, i in dc.items():\n",
        "   if i==ma: last=a\n",
        "\n",
        " return(max(dist))\n",
        "def step_reservoir_sampling(edge):\n",
        "    global init, w_i, w_1, M, window_counter, lamb, t_i, dict_test, tau, t\n",
        "    if init == 0:\n",
        "        # initialise w_i\n",
        "        w_1 = int(edge[2])+tau\n",
        "        w_i = int(edge[2])+tau\n",
        "        t_i = int(edge[2])+lamb\n",
        "        # window counter\n",
        "        window_counter = 1\n",
        "        # terminate initialisation\n",
        "        init+=1\n",
        "    # check if edge is indide w_i\n",
        "    if int(edge[2]) < t_i:\n",
        "        M[-1]+=1\n",
        "    if int(edge[2]) >= t_i:\n",
        "        print(M)\n",
        "        if int(edge[2]) >= w_1:\n",
        "          comp = components(reservoir)\n",
        "          comp1 = sorted(comp, reverse=True)\n",
        "          xx = comp_edges(comp1)\n",
        "          x = comp1[0][0]\n",
        "          y = comp1[0][1]\n",
        "          z = diameter(0)\n",
        "          t= window_counter\n",
        "          ll =[(x,y,z,t)]\n",
        "          if window_counter == 3:\n",
        "           x = comp1[0][0]\n",
        "           y = comp1[0][1]\n",
        "           z = diameter(0)\n",
        "           t= window_counter\n",
        "           print(\"window_counter  =\",t,\",\",\"vertices =\",x,\",\",\"edges=\", y,\",\",\"Diameter =\",z, end='\\n')\n",
        "          else:\n",
        "           x = comp1[0][0]\n",
        "           y = comp1[0][1]\n",
        "           z = diameter(0)\n",
        "           t= window_counter\n",
        "           print(t,x,y,z,sep=',', end='\\n')\n",
        "          #print(\"Diameter of largest component  =\",diameter(0))\n",
        "          now = datetime.datetime.now()\n",
        "          time_export = now.strftime(\"%Y_%m_%d_%Hh%M\")\n",
        "          #f = open(\"sample_data/%s_step_reservoir_edges.csv\" % (time_export), \"a\")\n",
        "          #f.write(\"Diameter,vertices,edges,time\\n\")\n",
        "          #f.close()\n",
        "          index = 0\n",
        "          for i in comp1:\n",
        "              if int(i[0]) >= int(threshold):\n",
        "                  #write_edge_reservoir(xx[index], time_export)\n",
        "                  #write_edge_reservoir(ll,time_export)\n",
        "                  index+=1\n",
        "              else:\n",
        "                  break\n",
        "        while len(reservoir) > 0 and int(reservoir[0][2]) < w_i-tau:\n",
        "            # while the previous condition is satisfy do\n",
        "            del reservoir[0]\n",
        "        window_counter +=1\n",
        "        del M[0]\n",
        "        M.append(1)\n",
        "        t_i += lamb\n",
        "        w_i += lamb\n",
        "    if sum(M) < k:\n",
        "        reservoir.append(edge)\n",
        "    else:\n",
        "        j = random.randint(0, sum(M))\n",
        "        if j < k:\n",
        "            if len(reservoir) < k:\n",
        "                reservoir.append(edge)\n",
        "            else:\n",
        "                del reservoir[j]\n",
        "                reservoir.append(edge)\n",
        "\n",
        "    return reservoir\n",
        "\n",
        "\n",
        "# Get date\n",
        "now = datetime.datetime.now()\n",
        "def ask_exit():\n",
        "    for task in asyncio.Task.all_tasks():\n",
        "        task.cancel()\n",
        "async def consume(queue):\n",
        "    while True:\n",
        "        # wait for an item from the producer\n",
        "        item = await queue.get()\n",
        "        twitter_edges_graph = create_graph(item)\n",
        "        queue.task_done()\n",
        "        global time_counter, global_counter, interval_counter\n",
        "        if len(twitter_edges_graph) > 0:\n",
        "            # process the item\n",
        "            # print('consuming {}...'.format(twitter_edges_graph))\n",
        "            # Notify the queue that the item has been processed\n",
        "            for edge in twitter_edges_graph:\n",
        "                if tracking[0] in edge[1].lower():\n",
        "                    pass\n",
        "                else:\n",
        "                    global_counter+=1\n",
        "                    interval_counter+=1\n",
        "                    window_reservoir_sampling = step_reservoir_sampling(edge)\n",
        "        if time.time()*1000 > start + timeout:\n",
        "            print('Capture terminated')\n",
        "            ask_exit()\n",
        "\n",
        "\n",
        "\n",
        "class Client(PeonyClient):\n",
        "    pass\n",
        "\n",
        "# every class inheriting from `PeonyClient` or `BasePeonyClient` has\n",
        "# an event_stream function that can be used on an `EventStream`\n",
        "@Client.event_stream\n",
        "class UserStream(EventStream):\n",
        "\n",
        "    def stream_request(self):\n",
        "        \"\"\"\n",
        "            The stream_request method returns the request\n",
        "            that will be used by the stream\n",
        "        \"\"\"\n",
        "        return self.stream.statuses.filter.post(track=tracking)\n",
        "\n",
        "\n",
        "    # the on_connect event is triggered on connection to an user stream\n",
        "    # https://dev.twitter.com/streaming/overview/messages-types#friends-lists-friends\n",
        "    @events.on_connect.handler\n",
        "    def connection(self, data):\n",
        "        consumer = asyncio.ensure_future(consume(queue))\n",
        "        print(\"Connected to stream!\")\n",
        "\n",
        "    # the on_tweet event is triggered when a tweet seems to be sent on\n",
        "    # the stream, by default retweets are included\n",
        "    @events.on_tweet.handler\n",
        "    async def tweet(self, data):\n",
        "        await queue.put(data)\n",
        "        #print(\"producing\")\n",
        "\n",
        "\n",
        "if __name__ == '__main__':\n",
        "    queue = asyncio.Queue()\n",
        "    client = Client(consumer_key=consumer_key,\n",
        "                     consumer_secret=consumer_secret,\n",
        "                     access_token=access_token,\n",
        "                     access_token_secret=access_token_secret)\n",
        "    client.run()"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}