Add basic implementation

laundry_notifier/run.py

@@ -0,0 +1,119 @@
+import time
+import datetime
+import smtplib
+import math
+from email.mime.text import MIMEText
+from datetime import datetime
+
+from microstacknode.hardware.accelerometer.mma8452q import MMA8452Q
+
+
+EMAIL_USERNAME = 'noreply@ianonavy.com'
+EMAIL_PASSWORD = 'S2PENjQbO6=cHgchw@CXs.bJ'
+RECIPIENT_EMAILS = [
+    'ianonavy@gmail.com', 'dmurvihill@gmail.com',
+    '5038304363@tmomail.net'
+]
+ALERT_EMAIL_TEXT = """Hi,
+
+Your laundry is done. Or maybe your roomate's. I don't know.
+
+Regretfully,
+Ianonavy Bot
+"""
+G_RANGE = 2
+INTERVAL = 0.005  # seconds
+WINDOW_SIZE = 40  # intervals
+THRESHOLD = 0.003  # G's
+MAX_EMAIL_FREQUENCY = 60  # seconds
+
+
+def average(s):
+    return sum(s) * 1.0 / len(s)
+
+
+def stdev(s):
+    if not s:
+        return 0
+    avg = average(list(s))
+    variance = map(lambda x: (x - avg) ** 2, s)
+    return math.sqrt(average(list(variance)))
+
+
+def notify_user(recipient_email_address):
+    print("Alerting " + recipient_email_address)
+
+    msg = MIMEText(ALERT_EMAIL_TEXT)
+
+    # me == the sender's email address
+    # you == the recipient's email address
+    msg['Subject'] = 'Laundry Finished'
+    msg['From'] = 'pi@localhost'
+    msg['To'] = recipient_email_address
+
+    # Send the message via our own SMTP server.
+    username = EMAIL_USERNAME
+    password = EMAIL_PASSWORD
+    server = smtplib.SMTP('smtp.gmail.com:587')
+    server.starttls()
+    server.login(username, password)
+    server.send_message(msg)
+    server.quit()
+
+
+def enqueue(sliding_window, item):
+    if len(sliding_window) == WINDOW_SIZE:
+        sliding_window.pop(0)
+    sliding_window.append(item)
+
+
+def amplitude_stdev(sliding_window):
+    standard_deviations = {}
+    for key in ('x', 'y', 'z'):
+        values = [instance[key] for instance in sliding_window]
+        standard_deviations[key] = stdev(values)
+    return standard_deviations
+
+
+def send_emails(last_email_sent_at):
+    print((datetime.now() - last_email_sent_at).seconds)  # for debugging
+    # limit frequency of emails
+    if (datetime.now() - last_email_sent_at).seconds > MAX_EMAIL_FREQUENCY:
+        [notify_user(email) for email in RECIPIENT_EMAILS]
+        return datetime.now()
+    return last_email_sent_at
+
+
+def main():
+    with MMA8452Q() as accelerometer:
+        # Configure accelerometer
+        accelerometer.standby()
+        accelerometer.set_g_range(G_RANGE)
+        accelerometer.activate()
+
+        # Settle
+        time.sleep(INTERVAL)
+
+        sliding_window = []
+        dryer_state = 'off'
+        last_email_sent_at = datetime(1970, 1, 1, 0, 0, 0)
+
+        while True:
+            g_values = accelerometer.get_xyz()
+            enqueue(sliding_window, g_values)
+            sliding_stdev = amplitude_stdev(sliding_window)
+            print(sliding_stdev, dryer_state)  # for debugging
+            # don't send emails right at the beginning
+            if g_values:
+                if sliding_stdev['x'] < THRESHOLD:
+                    if dryer_state == 'on':
+                        last_email_sent_at = send_emails(last_email_sent_at)
+                    dryer_state = 'off'
+                else:
+                    dryer_state = 'on'
+
+            time.sleep(INTERVAL)
+
+
+if __name__ == '__main__':
+    main()

setup.py

@@ -0,0 +1,17 @@
+from distutils.core import setup
+
+description = "Raspberry Pi-based laundry notification system."
+
+setup(
+    name='laundry-notifier',
+    version='0.0.1',
+    author='Ian Adam Naval',
+    author_email='ianonavy@gmail.com',
+    packages=['laundry_notifier'],
+    scripts=[],
+    url='https://git.ianonavy.com/ianonavy/laundry-notifier/',
+    license='MIT',
+    description=description,
+    long_description=description,
+    install_requires=['microstackcommon', 'microstacknode'],
+)