scale is working

.envrc

@@ -0,0 +1 @@
+use nix

README.md

@@ -0,0 +1,6 @@
+# coffee-scale
+
+## Build and flash
+```shell
+ampy -p /dev/ttyACM0 run main.py
+```

libs/hx711.py

@@ -0,0 +1,422 @@
+# type: ignore
+
+# MIT License
+# 
+# Copyright (c) 2022 Daniel Robertson
+# 
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+# 
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import _thread
+import time
+from machine import Pin
+from micropython import const
+from rp2 import PIO, StateMachine, asm_pio
+
+class hx711:
+
+    class _util:
+
+        @classmethod
+        def get_sm_from_pio(cls, pio: PIO, sm_index: int) -> StateMachine:
+            """Returns the StateMachine object from the given index
+
+            Args:
+                pio (PIO): RP2040 PIO instance
+                sm_index (int):
+
+            Returns:
+                StateMachine:
+            """
+            return pio.state_machine(sm_index)
+
+        @classmethod
+        def get_sm_index(cls, pio_offset: int, sm_offset: int) -> int:
+            """Returns the global state machine index from given args
+
+            Args:
+                pio_offset (int): 0 or 1
+                sm_offset (int):
+
+            Returns:
+                int: index between 0 and 7
+            """
+            return (pio_offset >> 2) + sm_offset
+
+        @classmethod
+        def get_pio_from_sm_index(cls, sm_index: int) -> PIO:
+            """Returns the correct PIO object from the global state machine index
+
+            Args:
+                sm_index (int):
+
+            Returns:
+                PIO:
+            """
+            return PIO(sm_index >> 2)
+
+        @classmethod
+        def sm_drain_tx_fifo(cls, sm: StateMachine) -> None:
+            """Clears the StateMachine TX FIFO
+
+            Args:
+                sm (StateMachine):
+            
+            Performs:
+            pull( ) noblock
+            https://github.com/raspberrypi/pico-sdk/blob/master/src/rp2_common/hardware_pio/pio.c#L252
+            This may not be thread safe
+            """
+            while sm.tx_fifo() != 0: sm.exec("pull() noblock")
+
+        @classmethod
+        def sm_get(cls, sm: StateMachine) -> int|None:
+            """Returns a value from the StateMachine's RX FIFO (NON-BLOCKING)
+
+            Args:
+                sm (StateMachine):
+
+            Returns:
+                int|None: None is returned if RX FIFO is empty
+            """
+            return sm.get() if sm.rx_fifo() != 0 else None
+
+        @classmethod
+        def sm_get_blocking(cls, sm: StateMachine) -> int:
+            """Returns a value from the StateMachine's RX FIFO (BLOCKING)
+
+            Args:
+                sm (StateMachine):
+
+            Returns:
+                int:
+            """
+            while sm.rx_fifo() == 0: pass
+            return sm.get()
+
+    class rate:
+        rate_10: int = const(0)
+        rate_80: int = const(1)
+
+    class gain:
+        gain_128: int = const(25)
+        gain_32: int = const(26)
+        gain_64: int = const(27)
+
+    class power:
+        pwr_up: int = const(0)
+        pwr_down: int = const(1)
+
+    class _pio_prog:
+        def __init__(self) -> None:
+            pass
+        def init(self, hx) -> None:
+            pass
+        def program(self) -> None:
+            pass
+
+    class pio_noblock(_pio_prog):
+
+        # see: https://github.com/endail/hx711-pico-c/blob/main/src/hx711_noblock.pio
+        PUSH_BITS: int = const(24)
+        FREQUENCY: int = const(10000000)
+
+        def __init__(self) -> None:
+            super().__init__()
+
+        def init(self, hx: hx711) -> None:
+            hx._sm = StateMachine(
+                hx._sm_index,
+                self.program,
+                freq=self.FREQUENCY,
+                in_base=hx.data_pin,
+                out_base=hx.clock_pin,
+                set_base=hx.clock_pin,
+                jmp_pin=None,
+                sideset_base=hx.clock_pin
+            )
+
+        # pylint: disable=E,W,C,R
+        @asm_pio(
+            out_init=(PIO.OUT_LOW),
+            set_init=(PIO.OUT_LOW),
+            sideset_init=(PIO.OUT_LOW),
+            out_shiftdir=PIO.SHIFT_LEFT,
+            autopush=True,
+            autopull=False,
+            push_thresh=PUSH_BITS,
+            fifo_join=PIO.JOIN_NONE
+        )
+        def program():
+
+            set(x, 0) # default gain of 0
+
+            label("wrap_target")
+            wrap_target()
+
+            set(y, 23) # read bits, 0 based
+
+            wait(0, pin, 0)
+
+            label("bitloop")
+            set(pins, 1)
+            in_(pins, 1)
+
+            jmp(y_dec, "bitloop").side(0).delay(2 - 1) # T4
+
+            pull(noblock).side(1)
+
+            out(x, 2)
+
+            jmp(not_x, "wrap_target").side(0)
+
+            mov(y, x)
+
+            label("gainloop")
+            set(pins, 1).delay(2 - 1) # T3
+            jmp(y_dec, "gainloop").side(0).delay(2 - 1) # T4
+
+            wrap()
+
+    READ_BITS: int = const(24)
+    MIN_VALUE: int = const(-0x800000)
+    MAX_VALUE: int = const(0x7fffff)
+    POWER_DOWN_TIMEOUT: int = const(60) # us
+    SETTLING_TIMES: list[int] = [ # ms
+        const(400),
+        const(50)
+    ]
+    SAMPLES_RATES: list[int] = [
+        const(10),
+        const(80)
+    ]
+
+    def __init__(
+        self,
+        clk: Pin,
+        dat: Pin,
+        sm_index: int = 0,
+        prog: _pio_prog = pio_noblock()
+    ):
+        """Create HX711 object
+
+        Args:
+            clk (Pin): GPIO pin connected to HX711's clock pin
+            dat (Pin): GPIO pin connected to HX711's data pin
+            sm_index (int, optional): Global state machine index to use. Defaults to 0.
+            prog (_pio_prog, optional): PIO program. Defaults to built-in pio_noblock().
+        """
+
+        self._mut = _thread.allocate_lock()
+        self._mut.acquire()
+
+        self.clock_pin: Pin = clk
+        self.data_pin: Pin = dat
+        self.clock_pin.init(mode=Pin.OUT)
+        self.data_pin.init(mode=Pin.IN)
+
+        self._sm: StateMachine
+        self._sm_index: int = sm_index
+        self._prog: __class__._pio_prog = prog
+
+        prog.init(self)
+
+        self._mut.release() 
+
+    def __bool__(self) -> bool:
+        return self._sm.active()
+
+    def __repr__(self) -> str:
+        return "[HX711 - CLK: {}, DAT: {}, SM_IDX: {}]".format(self.clock_pin, self.data_pin, self._sm_index)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, ex_type, ex_val, ex_tb) -> None:
+        # handle abrupt exits from locked contexts
+        if self._mut.locked(): self._mut.release()
+        self.close()
+
+    def close(self) -> None:
+        """Stop communication with HX711. Does not alter power state.
+        """
+        self._mut.acquire()
+        self._sm.active(0)
+        __class__._util.get_pio_from_sm_index(self._sm_index).remove_program(self._prog.program)
+        self._mut.release()
+
+    def set_gain(self, gain: int) -> None:
+        """Change HX711 gain
+
+        Args:
+            gain (int):
+        """
+        self._mut.acquire()
+        __class__._util.sm_drain_tx_fifo(self._sm)
+        self._sm.put(gain)
+        self._sm.get()
+        __class__._util.sm_get_blocking(self._sm)
+        self._mut.release()
+
+    @classmethod
+    def get_twos_comp(cls, raw: int) -> int:
+        """Returns the one's complement value from the raw HX711 value
+
+        Args:
+            raw (int): raw value from HX711
+
+        Returns:
+            int:
+        """
+        return -(raw & +cls.MIN_VALUE) + (raw & cls.MAX_VALUE)
+
+    @classmethod
+    def is_min_saturated(cls, val: int) -> bool:
+        """Whether value is at its maximum
+
+        Args:
+            val (int):
+
+        Returns:
+            bool:
+        """
+        return val == cls.MIN_VALUE
+
+    @classmethod
+    def is_max_saturated(cls, val: int) -> bool:
+        """Whether value is at its maximum
+
+        Args:
+            val (int):
+
+        Returns:
+            bool:
+        """
+        return val == cls.MAX_VALUE
+
+    @classmethod
+    def get_settling_time(cls, rate: int) -> int:
+        """Returns the appropriate settling time for the given rate
+
+        Args:
+            rate (int):
+
+        Returns:
+            int: milliseconds
+        """
+        return cls.SETTLING_TIMES[rate]
+
+    @classmethod
+    def get_rate_sps(cls, rate: int) -> int:
+        """Returns the numeric value of the given rate
+
+        Args:
+            rate (int):
+
+        Returns:
+            int:
+        """
+        return cls.SAMPLES_RATES[rate]
+
+    def get_value(self) -> int:
+        """Blocks until a value is returned
+
+        Returns:
+            int:
+        """
+        self._mut.acquire()
+        rawVal = __class__._util.sm_get_blocking(self._sm)
+        self._mut.release()
+        return self.get_twos_comp(rawVal)
+
+    def get_value_timeout(self, timeout: int = 1000000) -> int|None:
+        """Attempts to obtain a value within the timeout
+
+        Args:
+            timeout (int, optional): timeout in microseconds. Defaults to 1000000.
+
+        Returns:
+            int|None: None is returned if no value is obtained within the timeout period
+        """
+
+        endTime: int = time.ticks_us() + timeout
+        val: int|None = None
+
+        self._mut.acquire()
+
+        while(time.ticks_us() < endTime):
+            val = self._try_get_value()
+            if val != None: break
+
+        self._mut.release()
+
+        return self.get_twos_comp(val) if val else None
+
+    def get_value_noblock(self) -> int|None:
+        """Returns a value if one is available
+
+        Returns:
+            int|None: None is returned if no value is available
+        """
+        self._mut.acquire()
+        val = self._try_get_value()
+        self._mut.release()
+        return self.get_twos_comp(val) if val else None
+
+    def set_power(self, pwr: int) -> None:
+        """Changes the power state of the HX711 and starts/stops the PIO program
+
+        Args:
+            pwr (int):
+        """
+
+        self._mut.acquire()
+
+        if pwr == __class__.power.pwr_up:
+            self.clock_pin.low()
+            self._sm.restart()
+            self._sm.active(1)
+        elif pwr == __class__.power.pwr_down:
+            self._sm.active(0)
+            self.clock_pin.high()
+
+        self._mut.release()
+
+    @classmethod
+    def wait_settle(cls, rate: int) -> None:
+        """Waits for the appropriate amount of time for values to settle according to the given rate
+
+        Args:
+            rate (int):
+        """
+        time.sleep_ms(cls.get_settling_time(rate))
+
+    @classmethod
+    def wait_power_down(cls) -> None:
+        """Waits for the appropriate amount of time for the HX711 to power down
+        """
+        time.sleep_us(cls.POWER_DOWN_TIMEOUT)
+
+    def _try_get_value(self) -> int|None:
+        """Attempts to obtain a value if one is available
+
+        Returns:
+            int|None: None is returned if no value is available
+        """
+        words = __class__.READ_BITS / 8
+        return self._sm.get() if self._sm.rx_fifo() >= words else None

libs/ssd1306.py

@@ -0,0 +1,155 @@
+# MicroPython SSD1306 OLED driver, I2C and SPI interfaces
+
+from micropython import const
+import framebuf
+
+
+# register definitions
+SET_CONTRAST = const(0x81)
+SET_ENTIRE_ON = const(0xA4)
+SET_NORM_INV = const(0xA6)
+SET_DISP = const(0xAE)
+SET_MEM_ADDR = const(0x20)
+SET_COL_ADDR = const(0x21)
+SET_PAGE_ADDR = const(0x22)
+SET_DISP_START_LINE = const(0x40)
+SET_SEG_REMAP = const(0xA0)
+SET_MUX_RATIO = const(0xA8)
+SET_COM_OUT_DIR = const(0xC0)
+SET_DISP_OFFSET = const(0xD3)
+SET_COM_PIN_CFG = const(0xDA)
+SET_DISP_CLK_DIV = const(0xD5)
+SET_PRECHARGE = const(0xD9)
+SET_VCOM_DESEL = const(0xDB)
+SET_CHARGE_PUMP = const(0x8D)
+
+# Subclassing FrameBuffer provides support for graphics primitives
+# http://docs.micropython.org/en/latest/pyboard/library/framebuf.html
+class SSD1306(framebuf.FrameBuffer):
+    def __init__(self, width, height, external_vcc):
+        self.width = width
+        self.height = height
+        self.external_vcc = external_vcc
+        self.pages = self.height // 8
+        self.buffer = bytearray(self.pages * self.width)
+        super().__init__(self.buffer, self.width, self.height, framebuf.MONO_VLSB)
+        self.init_display()
+
+    def init_display(self):
+        for cmd in (
+            SET_DISP | 0x00,  # off
+            # address setting
+            SET_MEM_ADDR,
+            0x00,  # horizontal
+            # resolution and layout
+            SET_DISP_START_LINE | 0x00,
+            SET_SEG_REMAP | 0x01,  # column addr 127 mapped to SEG0
+            SET_MUX_RATIO,
+            self.height - 1,
+            SET_COM_OUT_DIR | 0x08,  # scan from COM[N] to COM0
+            SET_DISP_OFFSET,
+            0x00,
+            SET_COM_PIN_CFG,
+            0x02 if self.width > 2 * self.height else 0x12,
+            # timing and driving scheme
+            SET_DISP_CLK_DIV,
+            0x80,
+            SET_PRECHARGE,
+            0x22 if self.external_vcc else 0xF1,
+            SET_VCOM_DESEL,
+            0x30,  # 0.83*Vcc
+            # display
+            SET_CONTRAST,
+            0xFF,  # maximum
+            SET_ENTIRE_ON,  # output follows RAM contents
+            SET_NORM_INV,  # not inverted
+            # charge pump
+            SET_CHARGE_PUMP,
+            0x10 if self.external_vcc else 0x14,
+            SET_DISP | 0x01,
+        ):  # on
+            self.write_cmd(cmd)
+        self.fill(0)
+        self.show()
+
+    def poweroff(self):
+        self.write_cmd(SET_DISP | 0x00)
+
+    def poweron(self):
+        self.write_cmd(SET_DISP | 0x01)
+
+    def contrast(self, contrast):
+        self.write_cmd(SET_CONTRAST)
+        self.write_cmd(contrast)
+
+    def invert(self, invert):
+        self.write_cmd(SET_NORM_INV | (invert & 1))
+
+    def show(self):
+        x0 = 0
+        x1 = self.width - 1
+        if self.width == 64:
+            # displays with width of 64 pixels are shifted by 32
+            x0 += 32
+            x1 += 32
+        self.write_cmd(SET_COL_ADDR)
+        self.write_cmd(x0)
+        self.write_cmd(x1)
+        self.write_cmd(SET_PAGE_ADDR)
+        self.write_cmd(0)
+        self.write_cmd(self.pages - 1)
+        self.write_data(self.buffer)
+
+
+class SSD1306_I2C(SSD1306):
+    def __init__(self, width, height, i2c, addr=0x3C, external_vcc=False):
+        self.i2c = i2c
+        self.addr = addr
+        self.temp = bytearray(2)
+        self.write_list = [b"\x40", None]  # Co=0, D/C#=1
+        super().__init__(width, height, external_vcc)
+
+    def write_cmd(self, cmd):
+        self.temp[0] = 0x80  # Co=1, D/C#=0
+        self.temp[1] = cmd
+        self.i2c.writeto(self.addr, self.temp)
+
+    def write_data(self, buf):
+        self.write_list[1] = buf
+        self.i2c.writevto(self.addr, self.write_list)
+
+
+class SSD1306_SPI(SSD1306):
+    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
+        self.rate = 10 * 1024 * 1024
+        dc.init(dc.OUT, value=0)
+        res.init(res.OUT, value=0)
+        cs.init(cs.OUT, value=1)
+        self.spi = spi
+        self.dc = dc
+        self.res = res
+        self.cs = cs
+        import time
+
+        self.res(1)
+        time.sleep_ms(1)
+        self.res(0)
+        time.sleep_ms(10)
+        self.res(1)
+        super().__init__(width, height, external_vcc)
+
+    def write_cmd(self, cmd):
+        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
+        self.cs(1)
+        self.dc(0)
+        self.cs(0)
+        self.spi.write(bytearray([cmd]))
+        self.cs(1)
+
+    def write_data(self, buf):
+        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
+        self.cs(1)
+        self.dc(1)
+        self.cs(0)
+        self.spi.write(buf)
+        self.cs(1)

main.py

@@ -0,0 +1,109 @@
+import time
+
+import neopixel
+from machine import I2C, PWM, Pin, Timer
+
+import ssd1306
+from hx711 import *
+
+buzzer = PWM(Pin(8, Pin.OUT))
+neo = neopixel.NeoPixel(Pin(16), 1)
+button = Pin(7, Pin.IN, Pin.PULL_UP)
+i2c = I2C(1, scl=Pin(3), sda=Pin(2), freq=200000)
+display = ssd1306.SSD1306_I2C(128, 64, i2c)
+hx = hx711(Pin(14), Pin(15))
+scaling = 410
+value = 0
+tara = 0
+last_button = None
+
+
+# freq:hz, duration: ms
+def beep(freq, duration):
+    buzzer.freq(freq)
+    buzzer.duty_u16(22768)
+    time.sleep_ms(duration)
+    buzzer.duty_u16(0)
+
+
+def led(r, g, b):
+    neo[0] = (r, g, b)
+    neo.write()
+
+
+def callback(pin):
+    global last_button, tara
+    if not last_button or time.ticks_diff(time.ticks_ms(), last_button) > 500:
+        last_button = time.ticks_ms()
+    else:
+        print("Nope")
+        return
+
+    print("Interrupt has occured")
+    beep(1000, 100)
+    tara = value
+    print("Tara is {}".format(tara))
+
+
+smoothing = 3
+smoothing_values = []
+
+
+def read_scale(_):
+    global value, smoothing_values
+    val = hx.get_value()
+    corrected = val / scaling
+    smoothing_values.append(corrected)
+    smoothing_values = smoothing_values[-smoothing:]
+
+    value = int(sum(smoothing_values) / smoothing)
+
+
+def main():
+    global tara
+    print("Let's go!")
+    led(0, 0, 50)
+
+    display.text("Hello, World!", 0, 0, 1)
+    display.show()
+
+    # 2. power up
+    hx.set_power(hx711.power.pwr_up)
+
+    # 3. [OPTIONAL] set gain and save it to the hx711
+    # chip by powering down then back up
+    hx.set_gain(hx711.gain.gain_128)
+    hx.set_power(hx711.power.pwr_down)
+    hx711.wait_power_down()
+    hx.set_power(hx711.power.pwr_up)
+
+    # 4. wait for readings to settle
+    hx711.wait_settle(hx711.rate.rate_10)
+
+    display.text("Tara....", 0, 20, 1)
+    display.show()
+
+    for _ in range(3):
+        read_scale(None)
+
+    print("offset: {}".format(value))
+    tara = value
+
+    tim = Timer()
+    tim.init(period=100, mode=Timer.PERIODIC, callback=read_scale)
+
+    button.irq(trigger=Pin.IRQ_RISING, handler=callback)
+
+    while True:
+        x = value - tara
+        display.fill(0)
+        display.text("{} g".format(x), 0, 20, 1)
+        display.show()
+        time.sleep_ms(20)
+
+
+if __name__ == "__main__":
+    buzzer.init()
+    main()
+    hx.close()
+    buzzer.deinit()

shell.nix

@@ -0,0 +1,9 @@
+{ pkgs ? import <nixpkgs> {} }:
+
+pkgs.mkShell {
+  buildInputs = with pkgs; [
+    adafruit-ampy
+    micropython
+  ];
+
+}