store_monitor.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "from flask import Flask, jsonify, request, render_template, send_file\n",
    "import pandas as pd\n",
    "from datetime import datetime, timedelta\n",
    "import pytz\n",
    "from pytz import timezone\n",
    "import mysql.connector\n",
    "from mysql.connector import Error\n",
    "import csv, os, time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_connection():\n",
    "    try:\n",
    "        connection = mysql.connector.connect(\n",
    "            host=\"localhost\",\n",
    "            port=\"8889\",\n",
    "            user=\"ABC\",\n",
    "            password=\"test123\",\n",
    "            database=\"Work_DB\"\n",
    "        )\n",
    "        return connection\n",
    "    except Error as e:\n",
    "        print(\"Error while connecting to MySQL:\", e)\n",
    "        return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def check_new_records():\n",
    "    while(True):\n",
    "        connection = get_connection()\n",
    "        try:\n",
    "            cursor=connection.cursor()\n",
    "\n",
    "            #=====Menu hours.csv\n",
    "            with open('Menu hours.csv','r') as file:\n",
    "                csv_reader=csv.DictReader(file)\n",
    "                \n",
    "                cursor.execute('SELECT * from LK_menu_hours')\n",
    "                existing_records = cursor.fetchall()\n",
    "\n",
    "                '''\n",
    "                Since records are of varying types, we convert them all to string using list comprehension.\n",
    "                This enables us to compare rows from the csv with existing table records without worrying about their type. \n",
    "                '''\n",
    "                \n",
    "                existing_records = [tuple(str(value) for value in record) for record in existing_records]\n",
    "                \n",
    "\n",
    "                for row in csv_reader:\n",
    "                    record = tuple(row.values())\n",
    "                    \n",
    "                    if record not in existing_records: # This also ensures no duplicate rows are in the tables\n",
    "                        print(\"================ New record inserted in LK_menu_hours ================\")\n",
    "                        query = \"INSERT INTO LK_menu_hours VALUES (%s, %s, %s, %s)\"\n",
    "                        cursor.execute(query, record)\n",
    "                        connection.commit()\n",
    "                        print(record)\n",
    "                        \n",
    "            #=====bq-results.csv\n",
    "            with open('bq-results-20230125-202210-1674678181880.csv','r') as file:\n",
    "                csv_reader=csv.DictReader(file)\n",
    "                \n",
    "                cursor.execute('SELECT * from LK_bq_result')\n",
    "                existing_records = cursor.fetchall()\n",
    "                \n",
    "                existing_records = [tuple(str(value) for value in record) for record in existing_records]\n",
    "\n",
    "                for row in csv_reader:\n",
    "                    record = tuple(row.values())\n",
    "                    if record not in existing_records: \n",
    "                        print(\"================ New record inserted in LK_bq_result ================\")\n",
    "                        query = \"INSERT INTO LK_bq_result VALUES (%s, %s, %s, %s)\"\n",
    "                        cursor.execute(query, record)\n",
    "                        connection.commit()\n",
    "                        print(record)\n",
    "\n",
    "            #=====Menu hours.csv\n",
    "            with open('store status.csv','r') as file:\n",
    "                csv_reader=csv.DictReader(file)\n",
    "                \n",
    "                cursor.execute('SELECT * from LK_store_status')\n",
    "                existing_records = cursor.fetchall()\n",
    "                \n",
    "                existing_records = [tuple(str(value) for value in record) for record in existing_records]\n",
    "\n",
    "                for row in csv_reader:\n",
    "                    record = tuple(row.values())\n",
    "                    \n",
    "                    if record not in existing_records: \n",
    "                        print(\"================ New record inserted in LK_store_status ================\")\n",
    "                        query = \"INSERT INTO LK_store_status VALUES (%s, %s, %s, %s)\"\n",
    "                        cursor.execute(query, record)\n",
    "                        connection.commit()\n",
    "                        print(record)\n",
    "\n",
    "            cursor.close()\n",
    "            connection.close()\n",
    "        except Error as e:\n",
    "            print(e)\n",
    "        time.sleep(3600) #1 hour = 3600 sec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_stores(start_time, end_time):\n",
    "    connection = get_connection()\n",
    "    if connection:\n",
    "        try:\n",
    "            cursor = connection.cursor()\n",
    "            query = \"\"\"SELECT DISTINCT(store_id) FROM LK_store_status WHERE \n",
    "            timestamp_utc BETWEEN %s AND %s\"\"\"\n",
    "            cursor.execute(query, (str(start_time), str(end_time)))\n",
    "            stores=cursor.fetchall()\n",
    "            \n",
    "            #stores = [store[0] for store in cursor.fetchall()]\n",
    "            cursor.close()\n",
    "            connection.close()\n",
    "            return stores\n",
    "        except Error as e:\n",
    "            print(\"Error executing query (in get_stores):\", e)\n",
    "            return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_business_hours(store_id):\n",
    "    connection = get_connection()\n",
    "    if connection:\n",
    "        try:\n",
    "            cursor = connection.cursor()\n",
    "            query = \"SELECT day, start_time_local, end_time_local FROM LK_menu_hours WHERE store_id = %s\"\n",
    "            cursor.execute(query, (store_id,))\n",
    "            result=cursor.fetchall()\n",
    "            business_hours = {}\n",
    "            cursor.close()\n",
    "            connection.close()\n",
    "            if result:\n",
    "                \n",
    "                for row in result:\n",
    "                    day, start_time_local, end_time_local = row\n",
    "                    business_hours[day] = (start_time_local, end_time_local)\n",
    "            else:\n",
    "                for i in range(0,6):\n",
    "                    business_hours[i] = ('00:00:00', '23:59:59')\n",
    "                    \n",
    "            return business_hours\n",
    "        except Error as e:\n",
    "            print(\"Error executing query (in get_business_hours):\", e)\n",
    "            return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_timezone(store_id):\n",
    "    connection = get_connection()\n",
    "    if connection:\n",
    "        try:\n",
    "            cursor = connection.cursor()\n",
    "            query = \"SELECT timezone_str FROM LK_bq_result WHERE store_id = %s\"\n",
    "            cursor.execute(query, (store_id,))\n",
    "            result = cursor.fetchone()  #since we only have one timezone per store\n",
    "            cursor.close()\n",
    "            connection.close()\n",
    "            return result[0] if result else 'America/Chicago' #deafult timestamp: america/chicago\n",
    "        except Error as e:\n",
    "            print(\"Error executing query (in get_timezone):\", e)\n",
    "            return None\n",
    "            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_timestamp_in_business_hours(store_id, business_hours, timezn):\n",
    "    connection = get_connection()\n",
    "    if connection:\n",
    "        try:\n",
    "            cursor = connection.cursor()\n",
    "            query = \"SELECT timestamp_utc, status FROM LK_store_status WHERE store_id = %s ORDER BY timestamp_utc\"\n",
    "            cursor.execute(query, (str(store_id),))\n",
    "\n",
    "            status_range=[]\n",
    "            for row in cursor.fetchall():\n",
    "                timestamp_utc = row[0]\n",
    "                status=row[1]\n",
    "                timestamp_local = pd.to_datetime(timestamp_utc).astimezone(pytz.timezone(timezn)) #convert from utc to local timezone\n",
    "                timestamp_week=timestamp_local.weekday()\n",
    "\n",
    "                for i in business_hours.keys():\n",
    "                    if( timestamp_week == i and \n",
    "                    pd.to_datetime(business_hours[i][0]).time()<= timestamp_local.time()<= pd.to_datetime(business_hours[i][1]).time()):\n",
    "                        status_range.append((timestamp_local, status))\n",
    "            return status_range\n",
    "\n",
    "        except Error as e:\n",
    "            print(\"Error executing query (in get_timestamp_in_business_hours):\", e)\n",
    "            return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "def interpolate_status(store_id, max_timestamp, status_range):\n",
    "\n",
    "    #====interpolated range for 1 week\n",
    "     \n",
    "    end_time = max_timestamp\n",
    "    current_time = max_timestamp - timedelta(days=7)\n",
    "    interpolated_time_1w=[]\n",
    "    while ( current_time < end_time ): #not <= because if current_time = end_time will create duplicate\n",
    "        for i in range ( 0, len(status_range)-1 ): #-1 since we are checking against i+1 items\n",
    "            if(\\\n",
    "                status_range[i][0] <= current_time < status_range[i+1][0]\n",
    "                ):\n",
    "                interpolated_time_1w.append((current_time, status_range[i][1]))\n",
    "                break\n",
    "            else:\n",
    "                interpolated_time_1w.append((status_range[i+1][0], status_range[i+1][1]))\n",
    "                break\n",
    "            \n",
    "        current_time += timedelta(days=1)\n",
    "\n",
    "    #====interpolated range for 1 day\n",
    "    end_time = max_timestamp\n",
    "    current_time = max_timestamp - timedelta(days=1)\n",
    "    interpolated_time_1d=[]\n",
    "    while ( current_time < end_time ): #not <= because if current_time = end_time will create duplicate\n",
    "        for i in range ( 0, len(status_range)-1 ): #-1 since we are checking against i+1 items\n",
    "            if(\\\n",
    "                status_range[i][0] <= current_time < status_range[i+1][0]\n",
    "                ):\n",
    "                interpolated_time_1d.append((current_time, status_range[i][1]))\n",
    "                break\n",
    "            else:\n",
    "                interpolated_time_1d.append((status_range[i+1][0], status_range[i+1][1]))\n",
    "                break\n",
    "            \n",
    "        current_time += timedelta(hours=1)\n",
    "\n",
    "    #====interpolated range for 1 hour\n",
    "    end_time = max_timestamp\n",
    "    current_time = max_timestamp - timedelta(hours=1)\n",
    "    interpolated_time_1h=[]\n",
    "    while ( current_time < end_time ): #not <= because if current_time = end_time will create duplicate\n",
    "        for i in range ( 0, len(status_range)-1 ): #-1 since we are checking against i+1 items\n",
    "            if(\\\n",
    "                status_range[i][0] <= current_time < status_range[i+1][0]\n",
    "                ):\n",
    "                interpolated_time_1h.append((current_time, status_range[i][1]))\n",
    "                break\n",
    "            else:\n",
    "                interpolated_time_1h.append((status_range[i+1][0], status_range[i+1][1]))\n",
    "                break\n",
    "            \n",
    "        current_time += timedelta(minutes=1)\n",
    "\n",
    "\n",
    "    return interpolated_time_1w, interpolated_time_1d, interpolated_time_1h\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_uptime_downtime(interpolated_time_status):\n",
    "    uptime = 0\n",
    "    downtime = 0\n",
    "    \n",
    "    for current_time in interpolated_time_status:\n",
    "        if ( current_time[1] == 'active' ):\n",
    "            uptime += 1\n",
    "        else:\n",
    "            downtime += 1\n",
    "    return uptime, downtime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_report_id():\n",
    "    report_id = os.urandom(8).hex()\n",
    "    return report_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "#existing_report_id=[]\n",
    "\n",
    "global report_id_lst\n",
    "report_id_lst=['Generated Report IDs from Calling /trigger']\n",
    "\n",
    "global generating_csv\n",
    "generating_csv = True\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_report():\n",
    "    connection = get_connection()\n",
    "    if connection:\n",
    "        try:\n",
    "            cursor=connection.cursor()\n",
    "\n",
    "            #set max timestamp as current time\n",
    "            query = \"SELECT MAX(timestamp_utc) FROM LK_store_status\"\n",
    "            cursor.execute(query)\n",
    "            max_timestamp = cursor.fetchone()[0]\n",
    "            \n",
    "            max_timestamp=pd.to_datetime(max_timestamp)\n",
    "            start_time = max_timestamp - timedelta(days=7)\n",
    "            end_time = max_timestamp\n",
    "\n",
    "            #get store records within start_time and end_time\n",
    "            stores=get_stores(start_time, end_time)\n",
    "\n",
    "            while(True):    #generate new report_id as long as there exists a duplicate\n",
    "                #create csv file polled by report\n",
    "                report_id=generate_report_id()\n",
    "\n",
    "                if report_id not in report_id_lst: #checking if report_id is unique\n",
    "                    report_id_lst.append(report_id)\n",
    "                    csv_file = f'report_{report_id}.csv'\n",
    "                    fieldnames = [\n",
    "                        'store_id', \n",
    "                        'uptime_last_hour(minutes)', 'downtime_last_hour(minutes)',\n",
    "                        'uptime_last_day(hours)', 'downtime_last_day(hours)',\n",
    "                        'update_last_week(days)', 'downtime_last_week(days)'\n",
    "                        ]\n",
    "        \n",
    "                    #======end for loop\n",
    "                    global file\n",
    "                    with open(csv_file, mode='w', newline='') as file:\n",
    "                        writer=csv.DictWriter(file, fieldnames=fieldnames)\n",
    "                        #add headers for the csv file\n",
    "                        csv_headers={           \n",
    "                            'store_id':'store_id',\n",
    "                            'uptime_last_hour(minutes)':'uptime_last_hour(minutes)', \n",
    "                            'downtime_last_hour(minutes)':'downtime_last_hour(minutes)',\n",
    "                            'uptime_last_day(hours)':'uptime_last_day(hours)', \n",
    "                            'downtime_last_day(hours)':'downtime_last_day(hours)',\n",
    "                            'update_last_week(days)':'update_last_week(days)', \n",
    "                            'downtime_last_week(days)':'downtime_last_week(days)'\n",
    "                            }\n",
    "                        writer.writerow(csv_headers)\n",
    "\n",
    "                         #=====For each store_id in store\n",
    "                        for i in range(0, len(stores)):\n",
    "                            store_id = stores[i][0]\n",
    "                            #for testing\n",
    "                            #store_id='5704365908807155452'\n",
    "\n",
    "                            business_hours=get_business_hours(store_id)\n",
    "                            timezn = get_timezone(store_id)\n",
    "                            \n",
    "                            status_range = get_timestamp_in_business_hours(store_id, business_hours, timezn)\n",
    "                            #print(status_range[1])\n",
    "                            \n",
    "                            interpolated_time_status_1w, interpolated_time_status_1d, interpolated_time_status_1h = \\\n",
    "                                interpolate_status(store_id, max_timestamp, status_range)\n",
    "                            #print(interpolated_time_status[0])\n",
    "\n",
    "                            uptime_1w, downtime_1w = get_uptime_downtime(interpolated_time_status_1w)\n",
    "                            uptime_1d, downtime_1d = get_uptime_downtime(interpolated_time_status_1d)\n",
    "                            uptime_1h, downtime_1h = get_uptime_downtime(interpolated_time_status_1h)\n",
    "\n",
    "                            record_write={\n",
    "                                    'store_id': store_id, \n",
    "                                    'uptime_last_hour(minutes)':uptime_1h, \n",
    "                                    'downtime_last_hour(minutes)':downtime_1h,\n",
    "                                    'uptime_last_day(hours)': uptime_1d, \n",
    "                                    'downtime_last_day(hours)':downtime_1d,\n",
    "                                    'update_last_week(days)': uptime_1w, \n",
    "                                    'downtime_last_week(days)': downtime_1w\n",
    "                                }\n",
    "                            writer.writerow(record_write)\n",
    "                        global generating_csv \n",
    "                        generating_csv=False \n",
    "                    break\n",
    "                else:\n",
    "                    print(\"Record already exists...regenarating new id...\")\n",
    "            \n",
    "            \n",
    "            cursor.close()\n",
    "            connection.close()\n",
    "            return(report_id)\n",
    "        except Error as e:\n",
    "            print(\"SQL query error: \",e)\n",
    "            return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app=Flask(__name__, template_folder='template')\n",
    "\n",
    "global generating_csv\n",
    "generating_csv = True\n",
    "\n",
    "@app.route('/')  #landing\n",
    "def index():\n",
    "    global report_id_lst\n",
    "    return render_template('index.html', report_lst=report_id_lst)\n",
    "\n",
    "@app.route('/csv_update')  #check new records every hour\n",
    "def csv_update():\n",
    "    check_new_records()\n",
    "    return 'Check for new Records every hour'\n",
    "\n",
    "@app.route('/trigger_report')\n",
    "def trigger_report():       #trigger report generation\n",
    "    global generating_csv\n",
    "    generating_csv = True\n",
    "    report_id = generate_report()\n",
    "    return jsonify({'report_id': report_id})\n",
    "\n",
    "@app.route('/get_report/<report_id>', methods=['GET', 'POST'])\n",
    "def get_report(report_id):\n",
    "    filename = f'report_{report_id}.csv'\n",
    "    global generating_csv\n",
    "\n",
    "    # all reports have generated\n",
    "    if generating_csv == False and report_id in report_id_lst:\n",
    "        file_df=pd.read_csv(filename, nrows=10)\n",
    "        return render_template('status_complete.html', status='Complete', tables=file_df.to_html(), report_file=filename)\n",
    "    \n",
    "    #only last report is generating\n",
    "    if generating_csv == True and report_id in report_id_lst[:-1]:\n",
    "        file_df=pd.read_csv(filename, nrows=10)\n",
    "        return render_template('status_complete.html', status='Complete', tables=file_df.to_html(), report_file=filename)\n",
    "    \n",
    "    #check if last report had generated\n",
    "    if generating_csv==True:\n",
    "            return \"Running...\"\n",
    "\n",
    "@app.route('/download_csv/<filename>')\n",
    "def download_csv(filename):\n",
    "    #filename = f'report_{report_id}.csv'\n",
    "    return send_file(filename, as_attachment=True)\n",
    "\n",
    "if __name__==\"__main__\": \n",
    "    app.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.5 ('base')",
   "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.5"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "89417f56685fbe942a5af5267d517fa2457399d38fb42bb8b703dcc5eb403cea"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}