import json, requests from flask import Flask, request, session, render_template, redirect, url_for, Response, send_file from flask_session import Session import pandas as pd import numpy as np from matplotlib.figure import Figure import matplotlib.dates as mdates import base64 from tqdm import tqdm from io import BytesIO import logging logger = logging.getLogger() from functools import wraps from EagleEyev3 import * from settings import config SECRET_KEY = "this needs to be changed to something unique and not checked into git" # check if it could pull in a config object from settings.py if config: # start checking for keys in config object and set sensible defaults if 'days_of_history' in config: DAYS_OF_HISTORY = config['days_of_history'] else: # fallback to a sane default DAYS_OF_HISTORY = 1 if 'log_level' in config: logger.setLevel(config['log_level']) if 'client_secret' in config: SECRET_KEY = config['client_secret'] else: logging.setLevel(config['INFO']) logging.info(f"Using EagleEyev3 version {EagleEyev3.__version__}") app = Flask(__name__) app.secret_key = SECRET_KEY SESSION_TYPE = 'filesystem' SESSION_PERMANENT = False PERMANENT_SESSION_LIFETIME = 60 * 60 * 24 * 7 # one week in seconds app.config.from_object(__name__) Session(app) # $ flask routes -s rule # INFO:root:Using EagleEyev3 version 0.0.15 # Endpoint Methods Rule # --------------------- ------- ------------------------------- # index GET / # accounts GET /accounts # camera_detail GET /camera//events # camera_detail GET /camera//events/ # camera_live_preivew GET /camera//preview # camera__preivew_image GET /camera//preview_image # camera_status_plot GET /camera//status_plot # cameras GET /cameras # landing GET /landing # login_callback GET /login_callback # logout GET /logout # static GET /static/ # switch_account GET /switch_account def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): logging.debug(session) if 'een' in session and session['een'].access_token: logging.debug(f"@login_required access_token: {session['een'].access_token}") return f(*args, **kwargs) else: # failed to find a valid session so redirecting to landing page if 'een' in session: logging.debug(f"@login_required access_token: {session['een'].access_token}") else: logging.warn('@login_requried failed to find a valid session') return redirect(url_for('landing')) return decorated_function @app.route('/landing') def landing(): # sometimes you just want to get to the landing page and want to avoid the index redirect logic if 'een' in session: een = session['een'] else: een = EagleEyev3(config) session['een'] = een base_url = "https://auth.eagleeyenetworks.com/oauth2/authorize" path_url = f"?client_id={een.client_id}&response_type=code&scope=vms.all&redirect_uri={een.redirect_uri}" values = { "login_link": f"{base_url}{path_url}" } return render_template('cover.html', template_values=values) @app.route('/') @login_required def index(): # get een instance from session, @login_required already checked for it een = session['een'] # using current_user as a proxy for an established valid session if een.access_token == None: base_url = "https://auth.eagleeyenetworks.com/oauth2/authorize" path_url = f"?client_id={een.client_id}&response_type=code&scope=vms.all&redirect_uri={een.redirect_uri}" values = { "login_link": f"{base_url}{path_url}" } return render_template('cover.html', template_values=values) # call this to check if session is actually valid check = een.get_current_user() if 'success' not in check or check['success'] == False: base_url = "https://auth.eagleeyenetworks.com/oauth2/authorize" path_url = f"?client_id={een.client_id}&response_type=code&scope=vms.all&redirect_uri={een.redirect_uri}" values = { "login_link": f"{base_url}{path_url}" } return render_template('cover.html', template_values=values) else: logging.info(f"{check['success']} - check get_current_user {een.current_user['email']} {een.access_token}") een.get_list_of_cameras() een.get_list_of_accounts() if len(een.accounts) > 0 and een.active_account == None: # they need to pick and account logging.info(f"redirecting to accounts.html because they don't have an active account { een.active_account }") values = { "current_user": een.current_user, "accounts": een.accounts, "active_account": een.active_account } redirect(url_for('accounts')) values = { "current_user": een.current_user, "cameras": een.cameras, "camera_count": len(een.cameras), "camera_count_online": len([i for i in een.cameras if i.is_online()]), "camera_count_offline": len([i for i in een.cameras if i.is_offline()]), "accounts": een.accounts, "active_account": een.active_account } return render_template('index.html', template_values=values) @app.route('/login_callback') def login_callback(): # This is getting the ?code= querystring value from the HTTP request. code = request.args.get('code') # create a new een object and store it in their session een = EagleEyev3(config) session['een'] = een if (code): # use the include code parameter to complete login process oauth_object = een.login_tokens(code) logging.debug(oauth_object) # let's try resaving this to see if it fixs the missing access_token on first request after login session['een'] = een return redirect(url_for('index')) @app.route('/logout') def logout(): # logout isn't working APIv3 right now so don't wait for the call to fail before logging out # if 'een' in session: # een = session['een'] # een.logout() session.pop('een') return redirect(url_for('index')) @app.route('/cameras') @login_required def cameras(): # get een instance from session, @login_required already checked for it een = session['een'] logging.debug(een.get_list_of_cameras()) values = { "current_user": een.current_user, "cameras": een.cameras, "camera_count": len(een.cameras), "camera_count_online": len([i for i in een.cameras if i.is_online()]), "camera_count_offline": len([i for i in een.cameras if i.is_offline()]) } return render_template('cameras_partial.html', template_values=values) @app.route('/accounts') @login_required def accounts(): # get een instance from session, @login_required already checked for it een = session['een'] logging.debug(een.get_list_of_accounts()) values = { "current_user": een.current_user, "accounts": een.accounts, "active_account": een.active_account } return render_template('accounts_partial.html', template_values=values) @app.route('/switch_account') @login_required def switch_account(): # This is getting the ?account= querystring value from the HTTP request. account = request.args.get('account') # get een instance from session, @login_required already checked for it een = session['een'] if (account): # switch into account logging.info(f"attempting to switch into account {account}") logging.debug(een.login_from_reseller(target_account_id=account)) return redirect(url_for('index')) @app.route('/camera//preview_image') @login_required def camera__preivew_image(esn=None): # get een instance from session, @login_required already checked for it een = session['een'] camera = een.get_camera_by_id(esn) res = camera.get_live_preview() if res: return send_file(BytesIO(res.content), mimetype='image/jpeg') else: return send_file('static/placeholder.png') @app.route('/camera//preview') @login_required def camera_live_preivew(esn=None): # get een instance from session, @login_required already checked for it een = session['een'] camera = een.get_camera_by_id(esn) values = { "current_user": een.current_user, "camera": camera, "events": camera.events['status'] } return render_template('camera_preview.html', template_values=values) @app.route("/camera//events/") @app.route('/camera//events') @login_required def camera_detail(esn=None, days=DAYS_OF_HISTORY): # get een instance from session, @login_required already checked for it een = session['een'] camera = een.get_camera_by_id(esn) # because of API limitation, can only query 24 hours max for i in tqdm(range(0, days)): camera.get_list_of_events(end_timestamp=een.time_before(hours=24*i), \ start_timestamp=een.time_before(hours=24*(i+1))) values = { "current_user": een.current_user, "camera": camera, "events": camera.events['status'] } return render_template('camera_events_partial.html', template_values=values) @app.route('/camera//status_plot') @login_required def camera_status_plot(esn=None): # get een instance from session, @login_required already checked for it een = session['een'] cam = een.get_camera_by_id(esn) logging.debug(cam.events['status'][0]) if cam.events['status'][0]['endTimestamp'] == None: logging.debug('found empty end_timestamp') cam.events['status'][0]['endTimestamp'] = str(pd.Timestamp.utcnow()) atm_df = pd.DataFrame(cam.events['status'][::-1], columns=['id', 'startTimestamp', 'actorId', 'data']) atm_df['ts'] = pd.to_datetime(atm_df.startTimestamp) atm_df['status_desc'] = atm_df['data'].apply(lambda x: x[0]['newStatus']['connectionStatus']) atm_df['status'] = atm_df['status_desc'].replace(to_replace=['online', 'offline', 'error', 'deviceOffline', 'deviceOnline', 'off', 'bridgeOffline', 'unknown'], value=[1,0,0,0,0,0,0,0]) imp = atm_df.set_index(['ts']) imp['startTimestamp'] = pd.to_datetime(imp['startTimestamp']) imp = imp.drop(['id', 'actorId', 'data', 'status_desc'], axis=1) imp['status'] = imp['status'].ffill() data = imp.resample('S').ffill() # logging.debug(data.tail(200)) # data['status'] = data['status'].astype('int64') data = data.drop(['startTimestamp'], axis=1) # Generate the figure **without using pyplot**. fig = Figure(figsize=(6, 5), dpi=160) ax = fig.subplots() ax.step(data.index, data['status'], lw=2, color='blue') ax.set_title(cam.name) # ax.fill_between(data['startTimestamp'], data['status']) ax.axhline(1, color='green', lw=2, alpha=0.3) ax.axhline(0, color='red', lw=2, alpha=0.3) for label in ax.get_xticklabels(which='major'): label.set(rotation=30, horizontalalignment='right') # Save it to a temporary buffer. buf = BytesIO() fig.savefig(buf, format="png", transparent=True) # Embed the result in the html output. data = base64.b64encode(buf.getbuffer()).decode("ascii") return f"

Graph of Status Events

" if __name__ == '__main__': app.run(host=config.server_host, port=config.server_port)