Skeleton of keezer manager project

students/luke/project/README.md

@@ -0,0 +1,66 @@
+Python 220 Project: Keezer Management
+
+Purpose:
+
+Keep track of my beer fridge.
+
+Measurement
+ - temperature
+ - weight of components
+  - CO2/beer gas tanks
+  - kegs
+
+Display
+ - LCD screen
+  - Display pounds of CO2 remaining
+  - Display gallons of beer remaining per keg
+
+Alert
+ - send email when thresholds are reached
+  - tanks or kegs low
+  - temperature too high or low
+
+Retention
+ - store values for analysis
+  - graphing
+  - consumption projection
+
+
+
+
+Use case 1:
+Exhaust CO2 cannister, swap with a fresh one, verify refill on LCD.
+1. LCD reads 0 pounds CO2 remaining.
+2. Open keezer, disconnect CO2 lines, remove tank and regulator
+3. LCD reads negative value (tare weight of tank and regulator)
+4. Move regulator to new tank, reconnect lines, place in keezer
+5. LCD reads approximately 5 pounds CO2 remaining
+
+Use case 2:
+Serve a dozen beers.
+1. LCD reads 2 gallons remaining
+2. Dispense beers
+3. LCD reads 0.5 gallons remaining
+4. Email alert received warning of imminent keg pop
+
+
+Design:
+
+The application will run on a small device such as a raspberry pi.  Sensors connected via USB and/or GPIO inside the fridge will feed data to the application.  External displays connected via USB or HDMI will provide output.  Data will be logged over the network to a data sink, likely SQL but possibly an snmp trap sink.
+
+Classes:
+ - Driver
+  - Main application logic
+  - Scheduling
+
+ - Sensor
+  - Define interface for concrete sensor subclasses
+
+ - Display
+  - Accept sensor data
+  - Concrete subclasses will display or export data
+
+
+Deliverables:
+
+The initial implementation of the keezer project will use random data andflat text files for input and output using simple concrete subclasses.  Subsequent iterations will add additional subclasses for actual devices, protocols, etc.

students/luke/project/src/KeezerDisplay.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+
+
+class KeezerDisplay(object):
+    """ Output class for keezer management project """
+
+
+    def __init__(self):
+        """ Register self with driver """
+        pass
+
+
+    def ingest(self, datum):
+        """ Take data from driver """
+        pass
+
+
+    def display(self, data):
+        """ Show configured output """
+        return True
+
+
+if __name__ == "__main__":
+    outp = KeezerDisplay()
+    assert(outp.display())

students/luke/project/src/KeezerManager.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+
+"""
+Keezer management project.  See README.md
+"""
+
+import sys
+import os
+import imp
+import glob
+import time
+import logging
+import logging.handlers
+
+from KeezerSensor import KeezerSensor
+from KeezerDisplay import KeezerDisplay
+
+POLL_INTERVAL=10
+
+DFLFMT = "%(asctime)s %(filename)s:%(lineno)-3d %(levelname)s %(message)s"
+FORMATTER = logging.Formatter(DFLFMT)
+
+CONSOLE_HANDLER = logging.StreamHandler()
+CONSOLE_HANDLER.setLevel(logging.DEBUG)
+CONSOLE_HANDLER.setFormatter(FORMATTER)
+
+LOGGER = logging.getLogger()
+LOGGER.setLevel(logging.DEBUG)
+LOGGER.addHandler(CONSOLE_HANDLER)
+
+# Concrete sensor and display classes are loaded dynamically.  There are
+# a few approaches to this.  In addition to the goergevreilly one copied
+# below, there's another example at:
+# https://www.blog.pythonlibrary.org/2012/07/31/advanced-python-how-to-dynamically-load-modules-or-classes/
+
+# https://www.georgevreilly.com/blog/2016/03/02/PythonImportSubclassFromDynamicPath.html
+def import_class(implementation_filename, base_class):
+    """ Dynamically import class from filesystem using imp. """
+
+
+    impl_dir, impl_filename = os.path.split(implementation_filename)
+    module_name, _ = os.path.splitext(impl_filename)
+
+    try:
+        sys.path.insert(0, impl_dir)
+        fp, filename, description = imp.find_module(module_name)
+        module = imp.load_module(module_name, fp, filename, description)
+        logging.debug(f"trying to import fp {fp} "
+                      f" filename {filename} "
+                      f" description {description} ")
+        for name in dir(module):
+            logging.debug(f"name {name}")
+            obj = getattr(module, name)
+            logging.debug(f"obj {obj}")
+            try:
+                if (type(obj) == type(base_class)
+                        and issubclass(obj, base_class)
+                        and obj != base_class):
+                    return obj
+
+            except TypeError as excpt:
+                """ issubclass will throw TypeError for some imports """
+                logging.debug(f"caught {excpt}")
+
+        raise ValueError("No subclass of {0} in {1}".format(
+            base_class.__name__, implementation_filename))
+
+    finally:
+        sys.path.pop(0)
+
+
+class KeezerManager:
+    """
+    Driver class for keezer manager.
+
+    - Load sensor and display modules dynamically.
+    - Poll sensors.
+    - Pass sensor data to displays.
+    """
+
+    def __init__(self):
+        """ Load subclasses """
+        logging.debug("called")
+        self.sensors = []
+        self.displays = []
+
+        """ Populate sensors from sensors/ directory"""
+        for file in glob.glob("sensors/*.py"):
+            logging.debug(f"appending sensor: {file}")
+            self.sensors.append(import_class(file, KeezerSensor)())
+
+        """ Populate displays from displays/ directory"""
+        for file in glob.glob("displays/*.py"):
+            logging.debug(f"appending display: {file}")
+            self.displays.append(import_class(file, KeezerDisplay)())
+
+
+    def run(self):
+        """ Main loop: poll sensors and feed to displays """
+        logging.debug("called")
+        while True:
+            loop_start = time.clock()
+            logging.debug(f"loop started at {loop_start}")
+
+            for sensor in self.sensors:
+                data = sensor.poll()
+
+                for display in self.displays:
+                    display.display(data)
+
+            # sleep for remainder of polling interval
+            loop_end = time.clock()
+            logging.debug(f"loop ended at {loop_end}")
+            time_delta = loop_end - loop_start
+            sleep_for = POLL_INTERVAL - time_delta
+            if sleep_for < 0:
+                sleep_for = 0
+            logging.debug(f"sleeping for {sleep_for}")
+            time.sleep(sleep_for)
+
+
+if __name__ == "__main__":
+    # List registered sensors and displays
+    km = KeezerManager()
+    for sensor in km.sensors:
+        print(f"sensor: {sensor}")
+    for display in km.displays:
+        print(f"display: {display}")
+
+    km.run()

students/luke/project/src/KeezerSensor.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+
+
+class KeezerSensor(object):
+    """ Sensor class for keezer management project """
+
+
+    def __init__(self):
+        """ Register self with driver """
+        pass
+
+
+    def poll(self):
+        """ Return sensor reading """
+        return True
+
+
+if __name__ == "__main__":
+    sns = KeezerSensor()
+    assert(sns.poll())