6
\$\begingroup\$

Overview & Motivation

This project manages and displays sensor data (temperature, humidity, Air Quality Index). The sensor data is displayed by the sensor manufacturer, PurpleAir, but their tool is hard to navigate (e.g. can't look at data more than a few days old) and slow to load.

I've aimed to make a tool that addresses these issues and can be hosted on an individual's own website. Additional features include displaying forecast and current weather for a set location via the OpenWeather API, and displaying air quality warnings according to EPA thresholds.

This is how the web app looks right now:

enter image description here

This is my first large-ish, self-directed programming project. Any and all improvements welcome!

Usage

The app and database for data storage are deployed on Dokku. The user needs to push the code to Dokku; set up the app, environment variables, Postgres database, and host; and link a PurpleAir sensor. For full functionality, an OpenWeather API key should also be linked.

User preferences are set via environment variables in Dokku (e.g. latitude/longitude, display defaults). Display settings can be toggled in the app's GUI, which updates via callback.

The app uses Plotly, Dash, and Flask for the app management and dashboard. The Postgres database is managed through Python using psycopg2.

Contents

I've included here code for the main app (puts sensor data into database and makes the web app) and the app helper code (plots, data-fetching functions, etc).

The code not shown here includes:

  • Functions to calculate Air Quality Index based on pollutant concentrations
  • Class and methods to manage the database (e.g. inserting data, deleting tables/rows, checking existence of specific observation, etc)
  • Script to get user-defined environment variables or use defaults if not defined

Code

App

The app file establishes the Dash app object, which manages both the web app and data insertion, and database connections. There are two database connections:

  • One for writing data. This is saved as an attribute of an instance of the DB-managing class. Since all write queries use a single connection, they are done in serial. This could be slow, but since it happens in the background, it doesn't matter.
  • One for reading data. This is a connection pool so that data-fetching happens in parallel to decrease plot render time.

A write procedure is started when the URL receives a POST request from the PurpleAir sensor. This happens every 2 minutes if the sensor and WiFi are working correctly; the sensor data is processed and inserted into the relevant table.

Every time sensor data is received, the code makes a GET request to the OpenWeather API. The API only provides new data about every 15 minutes, so attempts to insert already-seen data are caught by UNIQUE constraints in the relevant tables and rolled back.

The app layout is defined here. The app shows three main plots, which can be manipulated by a series of selector tools (date range, temperature unit, etc). Manipulating a selector triggers one or more callbacks; callbacks are also automatically run every 2 minutes to display the most recent data, since sensor data is updated every 2 minutes. Callbacks for plots fetch data from the database and return formatted plot objects.

# -*- coding: utf-8 -*-

# Running app and building webpage.
import dash
import dash_core_components as dcc
import dash_html_components as html
from flask import Flask
from flask import request

# Making plots and handling data.
import plotly.graph_objects as go  # More complex plotly graphs
import pandas as pd
from requests import get  # Make get requests
import json  # Decode jsons
import page_helper as ph  # Functions to fetch data and build plots

# Managing database.
import psycopg2
from psycopg2 import extras
from psycopg2 import pool
import database_management as dm

import user_settings as us  # JSON header verification, API key, etc.


# Initializing the app and webpage.
external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css']

app = dash.Dash(__name__, external_stylesheets=external_stylesheets)
app.title = 'PurpleAir Monitoring'

server = app.server


# Get DB connection pool for fetching data.
connPool = pool.ThreadedConnectionPool(
    1, 10, us.databaseUrl, cursor_factory=extras.DictCursor)


# Get read and write DB connection for managing database. Initialize DB object.
writeConn = psycopg2.connect(us.databaseUrl)
db = dm.AirDatabase(writeConn)


# Add incoming data to DB.
@server.route('/sensordata', methods=['POST'])
def insert_data():
    if not db:
        raise Exception('db object not defined')

    if us.header_key and request.headers.get('X-Purpleair') == us.header_key:
        db.insert_sensor_row(request.json)
    elif not us.header_key:
        db.insert_sensor_row(request.json)

    if us.loadHistoricalData:
        # Add all historical data to DB.
        db.load_historal_data()

    if us.openWeatherApiKey:
        print('querying weather API')
        # Make get request to OpenWeather API.
        weatherResponse = get("https://api.openweathermap.org/data/2.5/onecall?lat={}&lon={}&appid={}&units=imperial&lang={}".format(
            us.latitude, us.longitude, us.openWeatherApiKey, us.lang))
        print('got weather API response')

        weatherData = json.loads(weatherResponse.content.decode('utf-8'))
        db.insert_weather_row_and_forecasts(weatherData)

    return 'done'



# Laying out the webpage.
forecastDisplaySettings = []

if us.showDailyForecast:
    forecastDisplaySettings.append('daily')
if us.showHourlyForecast:
    forecastDisplaySettings.append('hourly')


app.layout = html.Div(children=[

    html.Div([
        html.Div([
            html.Label('Select a date range to display:'
                       )], className="three columns"),
        html.Div([
            dcc.Dropdown(
                id='standard-date-picker',
                options=[
                    {'label': '1 day', 'value': '1 day'},
                    {'label': '3 days', 'value': '3 days'},
                    {'label': '1 week', 'value': '1 week'},
                    {'label': '2 weeks', 'value': '2 weeks'},
                    {'label': '1 month', 'value': '1 month'},
                    {'label': '6 months', 'value': '6 months'},
                    {'label': '1 year', 'value': '1 year'},
                    {'label': 'All time', 'value': 'all'},
                    {'label': 'Custom date range', 'value': 'custom'}
                ], value=us.defaultTimeRange
            )], className="three columns"),
        html.Div([
            dcc.DatePickerRange(
                id='custom-date-range-picker',
                start_date_placeholder_text='Select a date',
                end_date_placeholder_text='Select a date',
                disabled=True
            ),
            dcc.Interval(
                id='fetch-interval',
                interval=(2 * 60) * 1000,  # 2 minutes in milliseconds
                n_intervals=0
            )
        ], className="six columns")

    ], className="row"),


    html.Div([
        html.Div('Select forecast to display:', className="three columns"),
        html.Div([
            dcc.Checklist(
                options=[
                    {'label': 'Hourly forecast', 'value': 'hourly'},
                    {'label': 'Daily forecast', 'value': 'daily'}
                ],
                value=forecastDisplaySettings,
                id='forecast-picker'
            )], className="three columns"),
    ], className="row"),

    html.Div(
        html.H3('Forecast', id='forecast-heading'),
        className="row"),

    html.Div([
        html.Div(
            id='daily-forecast-boxes')
    ], className="row"),

    html.Div([
        html.Div(
            id='hourly-forecast-display')
    ], className="row"),

    html.Div([
        html.H3('Temperature')
    ], className="row"),

    # Plot of temperature. Dropdown to toggle between °F and °C.
    html.Div([
        html.Div([
            dcc.Graph(
                id='temp-vs-time',
            )], className="eight columns"),
        html.Div([
            html.Div(
                dcc.Dropdown(
                    id='temp-unit-picker',
                    options=[
                        {'label': '°F', 'value': 'temp_f'},
                        {'label': '°C', 'value': 'temp_c'}
                    ], value='temp_f'
                ), className="row"),
            html.Blockquote(
                id='curr-sensor-temp',
                className="row"),
            html.Blockquote(
                id='curr-outside-temp',
                className="row")
        ], className="three columns", style={'position': 'relative'}),
    ], className="row"),

    html.Div([
        html.H3('Humidity')
    ], className="row"),

    # Plot of humidity.
    html.Div([
        html.Div([
            dcc.Graph(
                id='humid-vs-time',
            )], className="eight columns"),
        html.Div([], className="four columns")
    ], className="row"),

    html.Div([
        html.H3('Air Quality Index')
    ], className="row"),

    # Plot of AQI (both PM 2.5 and 10.0). Multi-select dropdown to toggle between displaying one or both. Text display + color of associated warning message.
    html.Div([
        html.Div([
            dcc.Graph(
                id='aqi-vs-time',
            )], className="eight columns"),
        html.Div([
            html.Div([
                dcc.Dropdown(
                    id='aqi-picker',
                    options=[
                        {'label': 'PM 2.5', 'value': 'pm_2_5_aqi'},
                        {'label': 'PM 10.0', 'value': 'pm_10_0_aqi'}
                    ], value=['pm_2_5_aqi', 'pm_10_0_aqi'], multi=True
                )], className="row"),
            html.Blockquote(id='aqi-warning', className="row")
        ], className="three columns")
    ], className="row"),

])


# Webpage callbacks
# Toggle custom date range picker display setting only when date dropdown menu is set to custom.
@ app.callback(
    dash.dependencies.Output('custom-date-range-picker', 'disabled'),
    [dash.dependencies.Input('standard-date-picker', 'value')])
def displayCustomDateRangePicker(standardDate):
    if standardDate == 'custom':
        return False

    return True


# Regenerate temp vs time graph when inputs are changed.
@ app.callback(
    [dash.dependencies.Output('temp-vs-time', 'figure'),
     dash.dependencies.Output('curr-sensor-temp', 'children'),
     dash.dependencies.Output('curr-outside-temp', 'children')],
    [dash.dependencies.Input('standard-date-picker', 'value'),
     dash.dependencies.Input('custom-date-range-picker', 'start_date'),
     dash.dependencies.Input('custom-date-range-picker', 'end_date'),
     dash.dependencies.Input('temp-unit-picker', 'value'),
     dash.dependencies.Input('fetch-interval', 'n_intervals')])
def updateTempPlot(standardDate, customStart, customEnd, tempUnit, n):
    records = ph.fetchSensorData(connPool, tempUnit, standardDate, [
        customStart, customEnd])
    weather = ph.fetchWeatherDataNewTimeRange(connPool, tempUnit, standardDate, [
        customStart, customEnd])

    records = ph.correctTemp(records, tempUnit)

    fig = ph.temp_vs_time(records, tempUnit)
    fig.add_trace(go.Scattergl(x=weather.ts, y=weather[tempUnit],
                               mode='markers+lines', line={"color": "rgb(175,175,175)"},
                               hovertemplate='%{y:.1f}',
                               name='Official outside'))

    currentRecords = ph.fetchSensorData(connPool, tempUnit, '1 day')
    currentWeather = ph.fetchWeatherDataNewTimeRange(
        connPool, tempUnit, '1 day')

    currentRecords = ph.correctTemp(currentRecords, tempUnit)

    try:
        currSensorStatement = 'Current sensor temperature: {:.0f}°'.format(
            currentRecords.iloc[0][tempUnit])
        currWeatherStatement = 'Current outside temperature: {:.1f}°'.format(
            currentWeather.iloc[0][tempUnit])
    except IndexError as e:
        print(e)
        currSensorStatement = 'Current sensor temperature: Unknown'
        currWeatherStatement = 'Current outside temperature: Unknown'

    return fig, currSensorStatement, currWeatherStatement


# Regenerate humidity vs time graph when inputs are changed.
@ app.callback(
    dash.dependencies.Output('humid-vs-time', 'figure'),
    [dash.dependencies.Input('standard-date-picker', 'value'),
     dash.dependencies.Input('custom-date-range-picker', 'start_date'),
     dash.dependencies.Input('custom-date-range-picker', 'end_date'),
     dash.dependencies.Input('fetch-interval', 'n_intervals')])
def updateHumidPlot(standardDate, customStart, customEnd, n):
    records = ph.fetchSensorData(connPool, "humidity", standardDate, [
        customStart, customEnd])
    weather = ph.fetchWeatherDataNewTimeRange(connPool, "humidity", standardDate, [
        customStart, customEnd])

    records = ph.correctHumid(records)

    fig = ph.humid_vs_time(records)
    fig.add_trace(go.Scattergl(x=weather.ts, y=weather.humidity,
                               mode='markers+lines', line={"color": "rgb(175,175,175)"},
                               hovertemplate='%{y}',
                               name='Official outside'))

    return fig


# Regenerate AQI vs time graph when inputs are changed.
@ app.callback(
    [dash.dependencies.Output('aqi-vs-time', 'figure'), dash.dependencies.Output(
        'aqi-warning', 'children'), dash.dependencies.Output('aqi-warning', 'style')],
    [dash.dependencies.Input('standard-date-picker', 'value'),
     dash.dependencies.Input('custom-date-range-picker', 'start_date'),
     dash.dependencies.Input('custom-date-range-picker', 'end_date'),
     dash.dependencies.Input('aqi-picker', 'value'),
     dash.dependencies.Input('fetch-interval', 'n_intervals')])
def updateAqiPlot(standardDate, customStart, customEnd, aqiSpecies, n):
    if len(aqiSpecies) == 0:
        # Default to showing PM 2.5.
        aqiSpecies = ["pm_2_5_aqi"]

    records = ph.fetchSensorData(connPool, aqiSpecies, standardDate, [
        customStart, customEnd])

    warningMessage, style = ph.fetchAqiWarningInfo(
        connPool,
        aqiSpecies,
        standardDate,
        [customStart, customEnd])

    return ph.aqi_vs_time(records, aqiSpecies), warningMessage, style


# Generate daily forecast display with most recent data.
@ app.callback(
    [dash.dependencies.Output('forecast-heading', 'children'),
     dash.dependencies.Output('daily-forecast-boxes', 'children')],
    [dash.dependencies.Input('forecast-picker', 'value'),
     dash.dependencies.Input('temp-unit-picker', 'value'),
     dash.dependencies.Input('fetch-interval', 'n_intervals')])
def updateDailyForecast(forecastsToDisplay, tempUnit, n):
    if 'daily' not in forecastsToDisplay:
        if 'hourly' not in forecastsToDisplay:
            return [], []
        return 'Forecast', None

    tempSelector = {'temp_f': ['min_f', 'max_f'], 'temp_c': ['min_c', 'max_c']}
    degreeUnit = {'temp_f': '°F', 'temp_c': '°C'}
    columns = ['weather_type_id', 'short_weather_descrip', 'detail_weather_descrip',
               'weather_icon', 'precip_chance', 'uvi'] + tempSelector[tempUnit]

    records = ph.fetchDailyForecastData(connPool, columns)

    blockStyle = {
        'backgroundColor': 'rgba(223,231,244,1.0)',
        "width": "15%",
        "margin-left": '0.83333333333%',
        "margin-right": '0.83333333333%',
        "border-radius": 10}
    lineStyle = {
        "margin-left": 15,
        "margin-top": 0,
        "margin-bottom": 0}

    forecastBoxes = []

    # TODO: Not recommended to use iterrows(), though this dataframe is quite small.
    for index, row in records.iterrows():
        if index < 6:

            # Customize weather description by weather type. Weather type codes here: https://openweathermap.org/weather-conditions#Weather-Condition-Codes-2
            if round(row["weather_type_id"], -2) in (300, 700) or row["weather_type_id"] == 800:
                weatherDescription = row["short_weather_descrip"]
            elif round(row["weather_type_id"], -2) == 200 or (round(row["weather_type_id"], -2) == 800 and row["weather_type_id"] != 800):
                weatherDescription = row["detail_weather_descrip"]
            if round(row["weather_type_id"], -2) in (500, 600):
                weatherDescription = row["detail_weather_descrip"]

                # Swap "shower" and following word.
                weatherDescription = weatherDescription.split(' ')

                if 'shower' in weatherDescription:
                    swapIndex = weatherDescription.index('shower')
                    weatherDescription[swapIndex], weatherDescription[swapIndex +
                                                                      1] = weatherDescription[swapIndex + 1], weatherDescription[swapIndex]

                if round(row["weather_type_id"], -2) == 500:
                    # Drop any instances of "intensity"
                    weatherDescription = [
                        item for item in weatherDescription if item != "intensity"]

                weatherDescription = ' '.join(weatherDescription)

            weatherDescription = weatherDescription.capitalize()

            forecastBoxes.append(
                html.Div([
                    html.B([row['ts'].strftime('%B '), row['ts'].day,
                            html.Img(
                                src='http://openweathermap.org/img/wn/{}@2x.png'.format(
                                    row['weather_icon']),
                                style={'height': '25%',
                                       'width': '25%',
                                       'verticalAlign': 'middle'})],
                           style={"margin-left": 5}),
                    html.P([weatherDescription],
                           style=lineStyle),
                    html.P(["Min: ",
                            round(row[tempSelector[tempUnit][0]]),
                            degreeUnit[tempUnit]],
                           style=lineStyle),
                    html.P(["Max: ",
                            round(row[tempSelector[tempUnit][1]]),
                            degreeUnit[tempUnit]],
                           style=lineStyle),
                    html.P(["Chance of rain: ",
                            round(row['precip_chance'] * 100), '%'],
                           style=lineStyle),
                    html.P(["UV Index: ",
                            round(row['uvi'], 0)],
                           style=lineStyle)
                ], style=blockStyle,
                    className="two columns"))

    return 'Forecast', forecastBoxes


# TODO: Generate hourly forecast display.
@ app.callback(
    dash.dependencies.Output('hourly-forecast-display', 'children'),
    [dash.dependencies.Input('forecast-picker', 'value'),
     dash.dependencies.Input('temp-unit-picker', 'value'),
     dash.dependencies.Input('fetch-interval', 'n_intervals')])
def updateHourlyForecast(forecastsToDisplay, tempUnit, n):
    if 'hourly' not in forecastsToDisplay:
        return []

    return 'Hourly forecast display not yet implemented'

    tempSelector = {'temp_f': ['min_f', 'max_f'], 'temp_c': ['min_c', 'max_c']}
    degreeUnit = {'temp_f': '°F', 'temp_c': '°C'}


if __name__ == '__main__':
    app.run_server(debug=True)

App helper code

Each data-fetching function gets a connection and cursor from the connection pool object (provided as an argument). Each data-fetching function is table-specific. Columns to fetch are specified by name as a list of strings (varName; queryFields is intended to be used for processing and renaming table fields, e.g. 'old_field + 2 AS new_field').

The data is formatted as a pandas dataframe and returned. The cursor is closed and the connection is returned to connection pool.

correctTemp() and correctHumid() apply fixed corrections as reported by PurpleAir based on the company's calibration data.

Graphing functions take sensor data and display settings as arguments. Weather API data is added on as a separate trace in the respective callback. The AQI plot function includes stripes of color to reflect EPA AQI safety thresholds. These are added in a loop since I only want to display stripes up to the last one that the largest shown sensor value falls within.

# -*- coding: utf-8 -*-

import plotly.graph_objects as go  # More complex plotly graphs
import pandas as pd
import psycopg2

import user_settings as us


def fetchSensorData(pool, varName, standardDate=us.defaultTimeRange, customDate=None, queryFields=None, timezone=us.timezone):
    """
    Fetch updated data for a single variable or a list of variables when date range is changed.

    Args:
        varName: str or list of str corresponding to fields in the sensor_data table
        standardDate: str

    Returns:
        pandas dataframe of data fetched
    """
    conn = pool.getconn()
    conn.set_session(readonly=True)
    cur = conn.cursor()

    if isinstance(varName, str):
        varName = [varName]

    names = ['measurement_ts'] + varName

    if not queryFields:
        queryFields = ', '.join(names)
    else:
        if isinstance(queryFields, str):
            queryFields = [queryFields]

        queryFields = ', '.join(['measurement_ts'] + queryFields)

    records = None

    print("getting sensor data from database...")

    # Get data from database within desired time frame.
    if standardDate != 'custom':
        if standardDate == 'all':
            cur.execute(
                "SELECT {} FROM sensor_data ORDER BY measurement_ts DESC ".format(queryFields))
        else:
            cur.execute(
                "SELECT {} FROM sensor_data WHERE measurement_ts >= NOW() - INTERVAL '{}' ORDER BY measurement_ts DESC ".format(queryFields, standardDate))

    else:
        if customDate[0] and customDate[1]:
            cur.execute("SELECT {} FROM sensor_data WHERE measurement_ts >= '{}' and measurement_ts <= '{}' ORDER BY measurement_ts DESC ".format(
                queryFields, customDate[0], customDate[1]))
        else:
            records = pd.DataFrame(columns=names)

    # Format data.
    if not records:
        try:
            records = pd.DataFrame([{name: row[name] for name in names}
                                    for row in cur.fetchall()], columns=names)

            records.measurement_ts = records.measurement_ts.apply(
                lambda ts: ts.tz_convert(timezone))

        except psycopg2.ProgrammingError:
            print('no data in selected timeframe, creating empty dataframe')
            records = pd.DataFrame(columns=names)

    print("got data")

    cur.close()
    pool.putconn(conn)
    return records


def fetchAqiWarningInfo(pool, aqiSpecies=['pm_2_5_aqi', 'pm_10_0_aqi'], standardDate=us.defaultTimeRange, customDate=None):
    varNames = ['rgb', 'description', 'message']

    # AQI warning text and color.
    if "pm_2_5_aqi" in aqiSpecies and "pm_10_0_aqi" not in aqiSpecies:
        warningVars = ['pm_2_5_aqi_rgb as rgb',
                       'pm_2_5_aqi_description as description',
                       'pm_2_5_aqi_message as message']

    elif "pm_2_5_aqi" not in aqiSpecies and "pm_10_0_aqi" in aqiSpecies:
        warningVars = ['pm_10_0_aqi_rgb as rgb',
                       'pm_10_0_aqi_description as description',
                       'pm_10_0_aqi_message as message']

    elif "pm_2_5_aqi" in aqiSpecies and "pm_10_0_aqi" in aqiSpecies:
        warningVars = [
            'CASE WHEN pm_2_5_aqi >= pm_10_0_aqi THEN pm_2_5_aqi_rgb ELSE pm_10_0_aqi_rgb END AS rgb',
            'CASE WHEN pm_2_5_aqi >= pm_10_0_aqi THEN pm_2_5_aqi_description ELSE pm_10_0_aqi_description END AS description',
            'CASE WHEN pm_2_5_aqi >= pm_10_0_aqi THEN pm_2_5_aqi_message ELSE pm_10_0_aqi_message END AS message']

    else:
        warningVars = []
        varNames = []

    try:
        # First (most recent) row of warning info.
        warnings = fetchSensorData(pool,
                                   varNames, standardDate, customDate, warningVars).iloc[0]

        warningMessage = [warnings['description'], '.\r', warnings['message']]
        style = {
            'backgroundColor': warnings['rgb']
        }
    except IndexError:
        warningMessage = ''
        style = {}

    return warningMessage, style


def fetchWeatherDataNewTimeRange(pool, varName, standardDate=us.defaultTimeRange, customDate=None, timezone=us.timezone):
    """
    Fetch updated data for a single variable or a list of variables when date range is changed.

    Args:
        varName: str or list of str corresponding to fields in the weather_data table

    Returns:
        pandas dataframe of data fetched
    """
    conn = pool.getconn()
    conn.set_session(readonly=True)
    cur = conn.cursor()

    if isinstance(varName, str):
        varName = [varName]

    names = ['ts'] + varName
    queryFields = ', '.join(names)

    records = None

    print("getting weather data from database...")

    # Get data from database.
    if standardDate != 'custom':
        if standardDate == 'all':
            cur.execute(
                "SELECT {} FROM weather_data ORDER BY ts DESC ".format(queryFields))
        else:
            cur.execute(
                "SELECT {} FROM weather_data WHERE ts >= NOW() - INTERVAL '{}' ORDER BY ts DESC ".format(queryFields, standardDate))

    else:
        if customDate[0] and customDate[1]:
            cur.execute("SELECT {} FROM weather_data WHERE ts >= '{}' and ts <= '{}' ORDER BY ts DESC ".format(
                queryFields, customDate[0], customDate[1]))
        else:
            records = pd.DataFrame(columns=names)

    # Format data
    if not records:
        try:
            records = pd.DataFrame([{name: row[name] for name in names}
                                    for row in cur.fetchall()], columns=names)
            records.ts = records.ts.apply(
                lambda ts: ts.tz_convert(timezone))

        except psycopg2.ProgrammingError:
            print('no data in selected timeframe, creating empty dataframe')
            records = pd.DataFrame(columns=names)

    print("got data")

    cur.close()
    pool.putconn(conn)
    return records


def fetchForecastData(pool, varName, tableName, timezone=us.timezone):
    """
    Fetch all daily forecast data.

    Args:
        timezone:

    Returns:
        pandas dataframe of data fetched
    """
    conn = pool.getconn()
    conn.set_session(readonly=True)
    cur = conn.cursor()

    if isinstance(varName, str):
        varName = [varName]

    names = ['ts'] + varName
    queryFields = ', '.join(names)

    print("getting weather forecast from database...")

    # Get data from database.
    cur.execute(
        "SELECT {} FROM {} ORDER BY ts ASC ".format(queryFields, tableName))

    # Format data.
    try:
        records = pd.DataFrame([{name: row[name] for name in names}
                                for row in cur.fetchall()], columns=names)
        records.ts = records.ts.apply(
            lambda ts: ts.tz_convert(timezone))

    except psycopg2.ProgrammingError:
        print('no data in selected timeframe, creating empty dataframe')
        records = pd.DataFrame(columns=names)

    print('got data')

    cur.close()
    pool.putconn(conn)
    return records


def fetchDailyForecastData(pool, varName, timezone=us.timezone):
    return fetchForecastData(pool, varName, "daily_weather_forecast", timezone)


def fetchHourlyForecastData(pool, varName, timezone=us.timezone):
    return fetchForecastData(pool, varName, "hourly_weather_forecast", timezone)


def correctTemp(records, tempUnit):
    # Temp correction: https://de-de.facebook.com/groups/purpleair/permalink/722201454903597/?comment_id=722403448216731
    if tempUnit == "temp_c":
        records[tempUnit] = (
            (((records[tempUnit] * 9 / 5) + 32) - 8) - 32) * 5 / 9
    elif tempUnit == "temp_f":
        records[tempUnit] = records[tempUnit] - 8

    return records


def correctHumid(records):
    # Humidity correction: https://de-de.facebook.com/groups/purpleair/permalink/722201454903597/?comment_id=722403448216731
    records["humidity"] = records["humidity"] + 4

    return records



# Figures to insert.
defaultMargin = dict(b=100, t=0, r=0)


def temp_vs_time(records, species="temp_f", margin=defaultMargin):
    newTempLabel = {
        "temp_c": "Temperature [°C]", "temp_f": "Temperature [°F]"}[species]

    if records.empty:
        # Make empty/blank plot.
        records = pd.DataFrame(columns=["measurement_ts", "value"])
        species = "value"

    fig = go.Figure()

    fig.add_trace(go.Scattergl(x=records["measurement_ts"],
                               y=records[species],
                               mode='markers+lines',
                               hovertemplate='%{y:.0f}',
                               name='Sensor'))

    fig.update_layout(margin=margin,
                      hovermode="x",
                      legend=dict(
                          yanchor="top",
                          y=0.99,
                          xanchor="left",
                          x=0.01
                      ))
    fig.update_yaxes(title_text=newTempLabel)

    if not records.empty:
        xBounds = [min(records.measurement_ts),
                   max(records.measurement_ts)]
        fig.update_layout(xaxis_range=xBounds)

    return fig


def humid_vs_time(records, margin=defaultMargin):
    if records.empty:
        # Make empty/blank plot.
        records = pd.DataFrame(columns=["measurement_ts", "humidity"])

    fig = go.Figure()

    fig.add_trace(go.Scattergl(x=records["measurement_ts"],
                               y=records["humidity"],
                               mode='markers+lines',
                               hovertemplate='%{y}',
                               name='Sensor'))

    fig.update_layout(margin=margin,
                      hovermode="x",
                      legend=dict(
                          yanchor="top",
                          y=0.99,
                          xanchor="left",
                          x=0.01
                      ))
    fig.update_yaxes(title_text="Relative humidity [%]")

    if not records.empty:
        xBounds = [min(records.measurement_ts),
                   max(records.measurement_ts)]
        fig.update_layout(xaxis_range=xBounds)

    return fig


def aqi_vs_time(records, species=["pm_2_5_aqi", "pm_10_0_aqi"], margin=defaultMargin):
    if isinstance(species, str):
        species = [species]

    # Initialize figure
    fig = go.Figure()

    if not species or records.empty:
        # Make empty records df with correct column names.
        records = pd.DataFrame(columns=["measurement_ts"] + species)

    else:
        xBounds = [min(records.measurement_ts),
                   max(records.measurement_ts)]
        yBound = max(max(item for item in records[aqiType] if item is not None)
                     for aqiType in species)

        # EPA color bands by AQI risk.
        # TODO: pull from csv instead of hard-coding.
        colorCutoffs = [
            [50, 'rgba(0,228,0,0.3)'], [100, 'rgba(255,255,0,0.3)'],
            [150, 'rgba(255,126,0,0.3)'], [200, 'rgba(255,0,0,0.3)'],
            [300, 'rgba(143,63,151,0.3)'], [10000, 'rgba(126,0,35,0.3)']]
        colorList = list((item[1] for item in colorCutoffs))

        # Put AQI color band info into dataframe. Data should span min ts to max ts to get full coloring of plot area.
        colorCutoffs = [
            [bound] + cutoff for bound in xBounds for cutoff in colorCutoffs]
        colorCutoffs = pd.DataFrame(colorCutoffs, columns=[
                                    "measurement_ts", "aqi", "color"])

        # Add color stripe one at a time. Stop at the last AQI color band that includes the max AQI value seen in measured data.
        for index, color in enumerate(colorList):
            x = colorCutoffs.loc[colorCutoffs["color"]
                                 == color]["measurement_ts"]
            y = colorCutoffs.loc[colorCutoffs["color"] == color]["aqi"]

            fig.add_trace(go.Scatter(
                x=x, y=y,
                mode='lines',
                line=dict(width=0),
                fillcolor=color,
                fill='tozeroy' if index == 0 else 'tonexty',
                showlegend=False,
                hovertemplate=None,
                hoverinfo='skip'
            ))

            # Max AQI value within most recently added color band.
            if int(yBound) < y.iloc[0]:
                break

        # Set plot axes ranges.
        if index == len(colorCutoffs) - 1:
            # Cap y range at nearest hundred greater than max measured AQI value.
            fig.update_layout(
                yaxis_range=(0, round(yBound + 100, -2)),
                xaxis_range=xBounds
            )
        else:
            fig.update_layout(
                yaxis_range=(0, y.iloc[0]),
                xaxis_range=xBounds
            )

    # Add measured AQI values.
    aqiLabel = {"pm_2_5_aqi": "PM 2.5", "pm_10_0_aqi": "PM 10.0"}
    aqiColor = {"pm_2_5_aqi": "#636EFA", "pm_10_0_aqi": "#EF553B"}

    # Add measured series one by one.
    for aqiType in species:
        fig.add_trace(go.Scattergl(
            x=records["measurement_ts"], y=records[aqiType],
            mode="markers+lines",
            hovertemplate='%{y}',
            name=aqiLabel[aqiType],
            marker=dict(color=aqiColor[aqiType])
        ))

    fig.update_layout(
        legend=dict(
            yanchor="top",
            y=0.99,
            xanchor="left",
            x=0.01
        ),
        margin=margin,
        hovermode="x"
    )

    fig.update_yaxes(title_text="AQI")

    return fig

Any criticism or comments would be greatly appreciated!

\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

I have no experience with any of these libraries, so I can only comment on aspects of the language itself.


def insert_data():
    if not db:
        raise Exception('db object not defined')

Just to be clear, if db truly hasn't been assigned at this point, you cannot use if not db to check for that. Attempting to use a name before it's been associated with an object with raise a NameError, which you'd need to catch. At that point though, I'd probably just allow the original NameError to propagate and not worry about catching it unless there was other specific information that I wanted to add to the error.

If you meant that "db is in a bad, falsey state", I'd probably change the error message to be clearer what the problem is, and change raise Exception(. . . to raise ValueError(. . .. Throwing generic Exceptions isn't a great idea, since it makes it harder for the caller of the function to handle only specific errors.


if us.header_key and request.headers.get('X-Purpleair') == us.header_key:
    db.insert_sensor_row(request.json)
elif not us.header_key:
    db.insert_sensor_row(request.json)

It seems like this could be reduced down to:

if not us.header_key or request.headers.get('X-Purpleair') == us.header_key:
    db.insert_sensor_row(request.json)

If not us.header_key is false, the right operand of or will run, and you know at that point that us.header_key must be truthy.


def displayCustomDateRangePicker(standardDate):
    if standardDate == 'custom':
        return False

    return True

This can be just:

def displayCustomDateRangePicker(standardDate):
    return standardDate != 'custom'

Please using "snake_case" naming when naming function and variables.



Sorry, I'm more tired than I originally thought. Hopefully someone else can give you a more complete review.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.