Tissue Box implementation for Arduino with Processing.py

coordinate.py

class Point(tuple):
    def __new__(self, x, y):
       return tuple.__new__(Point, (x, y))

    @property
    def x(self):
        return self[0]

    @property
    def y(self):
        return self[1]

    def __add__(self, other):
        """
        returns: new point/ vector
        """
        if isinstance(other, Vector):
            return self.__class__(self.x+ other.x, self.y+other.y)
        else:
            raise TypeError("{self_type} can only be added by Vector, {other_type} provided".format(
                self_type = type(self).__name__, other_type = type(other).__name__))
    
    def __sub__(self, other):
        """
        returns: new point/ vector
        """
        if isinstance(other, Vector):
            return self.__class__(self.x - other.x, self.y - other.y)
        elif isinstance(other, self.__class__):
            return Vector(self.x - other.x, self.y - other.y)
        else:
            raise TypeError("{self_type} can only be substracted by Point or Vector, {other_type} provided".format(
                self_type = type(self).__name__, other_type = type(other).__name__))

    def __mul__(self, other):
        raise NotImplementedError("{self_type} cannot be multiplied")

    def __radd__(self, other):
        if other == 0:
            return self
        else:
            return self.__add__(other)
    
    def __str__(self):
        return "{t}({x}, {y})".format(t=type(self).__name__,x=self.x, y=self.y)

class Vector(Point):
    def __new__(self, x, y):
       return tuple.__new__(Vector, (x, y))
    
    @property
    def x(self):
        return self[0]

    @property
    def y(self):
        return self[1]

    def __mul__(self, other):
        if isinstance(other, (int, float)):
            return self.__class__(self.x * other, self.y * other)
        else:
            raise TypeError("{self_type} can only be multiplied by number, {other_type} provided".format(
                self_type = type(self).__name__, other_type = type(other).__name__))

    def sign(self, d=1):
        return self.__class__(
            self.__class__._sign(self.x, d=d), 
            self.__class__._sign(self.y, d=d)
            )

    def exchangeXY(self):
        return self.__class__(self.y, self.x)

    def flipBoth(self):
        return self.__class__(-self.x, -self.y)

    def flipX(self):
        return self.__class__(-self.x, self.y)

    def flipY(self):
        return self.__class__(self.x, -self.y)

    def split(self):
        return [self.__class__(self.x, 0),
                self.__class__(0, self.y)
               ]

    def length(self):
        return ((self.x ** 2) + (self.y ** 2)) ** 0.5

    def normalize(self):
        length = self.length()
        if length > 0:
            return self.__class__(self.x / length , self.y/ length)
        else:
            return self.__class__(0 , 0)
    
    def isZero(self):
        return self.x == self.y == 0

    @staticmethod
    def _sign(x, d):
        if x > 0:
            return d
        elif x < 0:
            return -d
        else:
            return 0

scheduler.py

import time

class Scheduler:
    all_schedulers = []

    def __init__(self, interval, callback,  until = 1e99, start_now = False, *args, **ngars):
        self.last_time = time.time()
        self.interval = interval
        self.until = until
        self.callback = (callback, args, ngars)
        self.exectutions = 0
        Scheduler.all_schedulers.append(self)
        if start_now:
            self.do()


    def do(self):
        (callback, args, ngars) = self.callback
        self.exectutions += 1
        self.last_time = time.time()
        # print(callback, Scheduler.all_schedulers, self.exectutions, self.until)
        if self.exectutions >= self.until:
            Scheduler.all_schedulers.remove(self)
        callback(*args, **ngars)
        
            

    def tick(self):
        if self.last_time + self.interval < time.time():
            self.do()

    @staticmethod
    def tick_all():
        [s.tick() for s in Scheduler.all_schedulers]
        

TissueBox.pyde

add_library("serial")
add_library("arduino")

from coordinate import *
from scheduler import Scheduler
# from entities import *

# TODO: move to seperate file
# ===  entities ===
class Button:
    all_buttons = []
    def __init__(self, pin, arduino):
        Button.all_buttons.append(self)
        self.pin = pin
        self.arduino = arduino
        self.arduino.pinMode(pin, Arduino.INPUT)
        self.last_pressed = False
        self.on_press_funcs = []
        self.on_release_funcs = []
    
    def __del__(self):
        Button.all_buttons.remove(self)

    def on_press(self, callback, *args, **ngars):
        self.on_press_funcs.append((callback, args, ngars))

    def on_release(self, callback, *args, **ngars):
        self.on_release_funcs.append((callback, args, ngars))

    def did_press(self):
        for (callback, args, ngars) in self.on_press_funcs:
            callback(*args, **ngars)
    
    def did_release(self):
        for (callback, args, ngars) in self.on_release_funcs:
            callback(*args, **ngars)

    def is_high(self):
        return self.arduino.digitalRead(self.pin) == Arduino.HIGH

    def detect_press(self):
        if self.last_pressed:
            if not self.is_high():
                self.last_pressed = False
                self.did_release()
        else: # last not pressed
            if self.is_high():
                self.last_pressed = True
                self.did_press()

    @staticmethod
    def update_status():
        [button.detect_press() for button in Button.all_buttons]

# class Reader:
#     def __init__(self, pin, arduino):
#         self.pin = pin
#         self.arduino = arduino
#         self.arduino.pinMode(pin, Arduino.INPUT)

class TissueBox:
    def __init__(self, read_pin):
        self.high = 1018
        self.low = 201
        self.pin = read_pin
        self.once_lows = []

    def reset_high(self):
        self.high = arduino.analogRead(self.pin) 
        return self.high 

    def reset_low(self):
        self.low = arduino.analogRead(self.pin) 
        return self.low 

    def read_volume(self):
        now = arduino.analogRead(self.pin) 
        volume = (1.0*now - self.low)/(self.high - self.low)
        for low in self.once_lows:
            (low_volume, activated, callback, args, ngars) = low
            if volume < low_volume:
                if activated:
                    self.once_lows.remove(low)
                    callback(*args, **ngars)
            elif not activated:
                self.once_lows.remove(low)
                self.once_lows.append((low_volume, True, callback, args, ngars))
        return volume

    def once_low(self, callback, low_volume = 0.2, start_immediently = False,  *args, **ngars):
        """
        """
        self.once_lows.append((low_volume, start_immediently, callback, args, ngars))



# ===============

def auto_detect_arduino_path():
    devices = [str(device) for device in Arduino.list()]
    selected = [device for device in devices if device.startswith('/dev/cu.usbmodem')]
    if len(selected) > 1:
        raise Exception("Found mutiple devices starts with `/dev/cu.usbmodem`"+str(selected))
    else:
        return selected[0]

    
def setup():
    global arduino
    size(1024, 768)

    reading_path = auto_detect_arduino_path()
    reading_band = 57600
    arduino = Arduino(this, reading_path, reading_band)
    print("Started listening arduino from "+str(reading_path)+":"+str(reading_band)+" ...")

    tissue_box = TissueBox(read_pin = 5)


    def set_high():
        high = tissue_box.reset_high()
        print("high set to: " + str(high))

    def set_low():
        low = tissue_box.reset_low()
        print("low set to: " + str(low))

    def once_low():
        print("Volume Low!")
        tissue_box.once_low(once_low)
        show_restocking()


    def show_restocking():
        if show_restocking not in registrations:
            registrations.append(show_restocking)
        stroke(0)
        fill(255,255,255,0.3*255)
        rect(1024/2-250, 768/2-200, 500, 400)

        f = createFont("Arial",16)
        textFont(f,32)      
        textAlign(CENTER)      
        fill(0)                       
        text("Re-ordering from amazon...",1024/2,768/2)

        def remove_restocking_message():
            if show_restocking  in registrations:
                registrations.remove(show_restocking)
        scheduler = Scheduler(interval = 3, callback = remove_restocking_message, until = 0)
        
        

    def read():
        volume = tissue_box.read_volume()
        volume_str = str(int(volume*100.0)) +"%"
        update_rect_params(volume)
        print("Volume: " + volume_str)
    def update_rect_params(volume):
        volume_str = str(int(volume*100.0)) +"%"
        rect_params["y"] = box_start_corner.y + (1.0 - volume) * box_size.y
        rect_params["height"] = volume * box_size.y
        rect_params["text"] = volume_str

    tissue_box.once_low(once_low)

    button_high = Button(pin = 2, arduino = arduino)
    button_high.on_press(set_high)
    button_low = Button(pin = 4, arduino = arduino)
    button_low.on_press(set_low)
    

    scheduler = Scheduler(interval = 1, callback = read)
    
    
    # originals
    box_start_corner = Point(89, 466)
    box_end_corner = Point(933, 680)
    box_size = box_end_corner - box_start_corner

    # dynamic
    rect_params = {
        "x" : box_start_corner.x,
        "y" : box_start_corner.y,
        "width" : box_size.x,
        "height" : box_size.y,
        'text' : '~'
    }
    
    def draw_img():
        img = loadImage("tissue.jpg")
        image(img,0,0)

        stroke(0)
        fill(66, 121, 176, 0.2 * 255)
        rect(rect_params['x'], rect_params['y'], rect_params['width'], rect_params['height'])

        f = createFont("Arial",16)
        textFont(f,32)      
        textAlign(CENTER)      
        fill(255)                       
        text(rect_params['text'],1024/2,rect_params['y'] + rect_params['height'] - 32)

    registrations.append(draw_img)

registrations = []
def draw():
    Button.update_status()
    Scheduler.tick_all()

    background(255)

    for callback in registrations:
        callback()
    

In short

def setup():

    def set_high():
        # set high value
        ...
    # set pin 2 as button mode
    button_high = Button(pin = 2, arduino = arduino)
    button_high.on_press(set_high)

    def read():
        # read volumes
    # read every 1 second
    scheduler = Scheduler(interval = 1, callback = read)

    def once_low():
        # call amazon
    tissue_box = TissueBox(read_pin = 5)
    tissue_box.once_low(once_low)