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 EagleEyev3 import * from settings import config logging.info(f"Using EagleEyev3 version {EagleEyev3.__version__}") app = Flask(__name__) SESSION_TYPE = 'filesystem' app.config.from_object(__name__) Session(app) # 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 else: if 'log_level' in config: logger.setLevel(config['log_level']) else: logging.setLevel(config['INFO']) @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('/') def index(): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) session['een'] = 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() values = { "current_user": een.current_user, "cameras": een.cameras, "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') if 'een' in session: een = session['een'] else: een = EagleEyev3(config) if (code): # use the include code parameter to complete login process oauth_object = een.login_tokens(code) return redirect(url_for('index')) @app.route('/logout') def logout(): if 'een' in session: een = session['een'] een.logout() session.pop('een') return redirect(url_for('index')) @app.route('/cameras') def cameras(): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) een.get_list_of_cameras() values = { "current_user": een.current_user, "cameras": een.cameras } return render_template('cameras_partial.html', template_values=values) @app.route('/accounts') def accounts(): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) 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') def switch_account(): # This is getting the ?account= querystring value from the HTTP request. account = request.args.get('account') if 'een' in session: een = session['een'] else: een = EagleEyev3(config) if (account): # use the include code parameter to complete login process een.login_from_reseller(target_account_id=account) return redirect(url_for('index')) @app.route('/camera//preview_image') def camera__preivew_image(esn=None): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) 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') def camera_live_preivew(esn=None): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) 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') def camera_detail(esn=None, days=DAYS_OF_HISTORY): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) camera = een.get_camera_by_id(esn) now = een.time_now() # 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(ts=now, hours=24*i), \ start_timestamp=een.time_before(ts=now, 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') def camera_status_plot(esn=None): if 'een' in session: een = session['een'] else: een = EagleEyev3(config) cam = een.get_camera_by_id(esn) 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'], value=[1,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) data = imp.resample('S').bfill() 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=een.server_host, port=een.server_port)