python cleanup

remove unaccessed imports

src/AirQualityMonitor.py

@@ -1,45 +0,0 @@
-import json
-import os
-import time
-import datetime
-import serial
-import redis
-import aqi
-
-redis_client = redis.StrictRedis(host=os.environ.get('REDIS_HOST'), port=6379, db=0)
-
-
-class AirQualityMonitor():
-
-    def __init__(self):
-        self.ser = serial.Serial(os.environ.get('SERIAL_DEVICE','/dev/ttyUSB0'))
-
-    def get_measurement(self):
-        self.data = []
-        for index in range(0,10):
-            datum = self.ser.read()
-            self.data.append(datum)
-        self.pmtwo = int.from_bytes(b''.join(self.data[2:4]), byteorder='little') / 10
-        self.pmten = int.from_bytes(b''.join(self.data[4:6]), byteorder='little') / 10
-        myaqi = aqi.to_aqi([(aqi.POLLUTANT_PM25, str(self.pmtwo)),
-                            (aqi.POLLUTANT_PM10, str(self.pmten))])
-        self.aqi = float(myaqi)
-
-        self.meas = {
-            "timestamp": datetime.datetime.now(),
-            "pm2.5": self.pmtwo,
-            "pm10": self.pmten,
-            "aqi": self.aqi,
-        }
-        return {
-            'time': int(time.time()),
-            'measurement': self.meas
-        }
-
-    def save_measurement_to_redis(self):
-        """Saves measurement to redis db"""
-        redis_client.lpush('measurements', json.dumps(self.get_measurement(), default=str))
-
-    def get_last_n_measurements(self):
-        """Returns the last n measurements in the list"""
-        return [json.loads(x) for x in redis_client.lrange('measurements', 0, -1)]

src/app.py

@@ -1,87 +1,90 @@
 import os
 import time
-from flask import Flask, request, jsonify, render_template
-from AirQualityMonitor import AirQualityMonitor
-from apscheduler.schedulers.background import BackgroundScheduler
-import redis
 import atexit
+
+from flask import Flask, jsonify, render_template
+from monitor import AirQualityMonitor
+from apscheduler.schedulers.background import BackgroundScheduler
 from flask_cors import CORS, cross_origin
 
+# initialize Flask and CORS
 app = Flask(__name__)
 cors = CORS(app)
-app.config['CORS_HEADERS'] = 'Content-Type'
-aqm = AirQualityMonitor()
+app.config["CORS_HEADERS"] = "Content-Type"
 
+# initialize AirQualityMonitor and scheduler
+aqm = AirQualityMonitor()
 scheduler = BackgroundScheduler()
 scheduler.add_job(func=aqm.save_measurement_to_redis, trigger="interval", seconds=60)
 scheduler.start()
-atexit.register(lambda: scheduler.shutdown())
+
 
 def pretty_timestamps(measurement):
-	timestamps = []
-	for x in measurement:
-		timestamp = x['measurement']['timestamp']
-		timestamps += [timestamp.split('.')[0]]
-	return timestamps
+    """Convert timestamps to a more readable format."""
+    return [x["measurement"]["timestamp"].split(".")[0] for x in measurement]
+
 
 def reconfigure_data(measurement):
     """Reconfigures data for chart.js"""
-    current = int(time.time())
-    measurement = measurement[:30]
-    measurement.reverse()
+    measurement = measurement[:30][::-1]
     return {
-        'labels': pretty_timestamps(measurement),
-        'aqi': {
-            'label': 'aqi',
-            'data': [x['measurement']['aqi'] for x in measurement],
-            'backgroundColor': '#181d27',
-            'borderColor': '#181d27',
-            'borderWidth': 3,
+        "labels": pretty_timestamps(measurement),
+        "aqi": {
+            "label": "aqi",
+            "data": [x["measurement"]["aqi"] for x in measurement],
+            "backgroundColor": "#181d27",
+            "borderColor": "#181d27",
+            "borderWidth": 3,
         },
-        'pm10': {
-            'label': 'pm10',
-            'data': [x['measurement']['pm10'] for x in measurement],
-            'backgroundColor': '#cc0000',
-            'borderColor': '#cc0000',
-            'borderWidth': 3,
+        "pm10": {
+            "label": "pm10",
+            "data": [x["measurement"]["pm10"] for x in measurement],
+            "backgroundColor": "#cc0000",
+            "borderColor": "#cc0000",
+            "borderWidth": 3,
         },
-        'pm2': {
-            'label': 'pm2.5',
-            'data': [x['measurement']['pm2.5'] for x in measurement],
-            'backgroundColor': '#42C0FB',
-            'borderColor': '#42C0FB',
-            'borderWidth': 3,
+        "pm2": {
+            "label": "pm2.5",
+            "data": [x["measurement"]["pm2.5"] for x in measurement],
+            "backgroundColor": "#42C0FB",
+            "borderColor": "#42C0FB",
+            "borderWidth": 3,
         },
     }
 
-@app.route('/')
+
+@app.route("/")
 def index():
     """Index page for the application"""
     context = {
-        'historical': reconfigure_data(aqm.get_last_n_measurements()),
+        "historical": reconfigure_data(aqm.get_last_n_measurements()),
     }
-    return render_template('index.html', context=context)
+    return render_template("index.html", context=context)
 
 
-@app.route('/api/')
+@app.route("/api/")
 @cross_origin()
-
 def api():
     """Returns historical data from the sensor"""
     context = {
-        'historical': reconfigure_data(aqm.get_last_n_measurements()),
+        "historical": reconfigure_data(aqm.get_last_n_measurements()),
     }
     return jsonify(context)
 
 
-@app.route('/api/now/')
+@app.route("/api/now/")
 def api_now():
     """Returns latest data from the sensor"""
     context = {
-        'current': aqm.get_measurement(),
+        "current": aqm.get_measurement(),
     }
     return jsonify(context)
 
 
 if __name__ == "__main__":
-    app.run(debug=True, use_reloader=False, host='0.0.0.0', port=int(os.environ.get('PORT', '8000')))
+    app.run(
+        debug=True,
+        use_reloader=False,
+        host="0.0.0.0",
+        port=int(os.environ.get("PORT", "8000")),
+    )

src/monitor.py

@@ -0,0 +1,50 @@
+import json
+import os
+import time
+import datetime
+import serial
+import redis
+import aqi
+
+
+REDIS_HOST = os.environ.get("REDIS_HOST", "localhost")
+REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379))
+REDIS_DB = int(os.environ.get("REDIS_DB", 0))
+redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
+
+
+class AirQualityMonitor:
+    SERIAL_DEVICE = os.environ.get("SERIAL_DEVICE", "/dev/ttyUSB0")
+
+    def __init__(self):
+        self.ser = serial.Serial(self.SERIAL_DEVICE)
+
+    def get_measurement(self):
+        """Fetches a measurement from the sensor and returns it."""
+        data = [self.ser.read() for _ in range(10)]
+        pmtwo = int.from_bytes(b"".join(data[2:4]), byteorder="little") / 10
+        pmten = int.from_bytes(b"".join(data[4:6]), byteorder="little") / 10
+        aqi_value = aqi.to_aqi(
+            [
+                (aqi.POLLUTANT_PM25, str(pmtwo)),
+                (aqi.POLLUTANT_PM10, str(pmten)),
+            ]
+        )
+
+        measurement = {
+            "timestamp": datetime.datetime.now(),
+            "pm2.5": pmtwo,
+            "pm10": pmten,
+            "aqi": float(aqi_value),
+        }
+        return {"time": int(time.time()), "measurement": measurement}
+
+    def save_measurement_to_redis(self):
+        """Saves measurement to redis db"""
+        redis_client.lpush(
+            "measurements", json.dumps(self.get_measurement(), default=str)
+        )
+
+    def get_last_n_measurements(self):
+        """Returns the last n measurements in the list"""
+        return [json.loads(x) for x in redis_client.lrange("measurements", 0, -1)]