init

filter_dev/app.py

@@ -0,0 +1,91 @@
+import tkinter as tk
+from tkinter import ttk
+import matplotlib.pyplot as plt
+from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
+
+from .filter import *
+from .gui.device import Device
+
+class FilterDevApp:
+    def __init__(self, root):
+        self.filter = None
+
+        self.root = root
+        self.root.title("JannTers Filter Evaluation Tool")
+
+        # Create a frame for the plot and sliders
+        self.frame = tk.Frame(self.root)
+        self.frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
+
+        # Create a figure for plotting
+        self.fig, self.ax = plt.subplots()
+        self.canvas = FigureCanvasTkAgg(self.fig, master=self.frame)
+        self.canvas.get_tk_widget().pack(side=tk.TOP, fill=tk.BOTH, expand=True)
+
+        # Create a frame for sliders
+        self.toolbar = tk.Frame(self.root)
+        self.toolbar.pack(side=tk.RIGHT, fill=tk.Y)
+
+        # Device Settings
+        self.device_label = ttk.Label(self.toolbar, text="Device Name:")
+        self.device_label.pack(pady=10)
+        self.device_name = ttk.Entry(self.toolbar)
+        self.device_name.insert(0, "Smaage")  # Set default value
+        self.device_name.pack(pady=10)
+
+        self.device = Device(self.device_name)
+        
+        self.connect_button = ttk.Button(self.toolbar, text="Connect Device", command=self.device.connect)
+        self.connect_button.pack(pady=10)
+
+        # Filter Settings
+        self.filter_type_label = ttk.Label(self.toolbar, text="Filter:")
+        self.filter_type_label.pack(pady=10)
+        self.filter_type_combobox = ttk.Combobox(self.toolbar, values=["MovAvg"])
+        self.filter_type_combobox.set("MovAvg")  # Set default value
+        self.filter_type_combobox.pack(pady=10)
+        self.change_filter = ttk.Button(self.toolbar, text="Change Filter", command=self.update_filter)
+        self.change_filter.pack(pady=10)
+        
+        # Objects
+        self.filter = MovAvg(self.device, self.toolbar, self.update_plot)
+
+        # Initial plot
+        self.update_plot()
+
+    def update_plot(self, *args):
+        if self.filter is None:
+            return
+
+        # Clear the current plot
+        self.ax.clear()
+
+        # Get current values from sliders
+        df = self.filter()
+
+        # Generate data
+        x = df['timestamps']
+        y1 = df['weights']
+        y2 = df['filtered']
+
+        # Plot the data
+        self.ax.plot(x, y1)
+        self.ax.plot(x, y2)
+        self.ax.set_xlabel("Time in ms")
+        self.ax.set_ylabel("Weight")
+        self.ax.grid()
+
+        # Draw the updated plot
+        self.canvas.draw()
+    
+    def update_filter(self):
+        option = self.filter_type_combobox.get()
+        if option == 'MovAvg' and not isinstance(self.filter, MovAvg):
+            self.filter = MovAvg(self.device, self.toolbar, self.update_plot)
+        
+        self.update_plot()
+
+if __name__ == "__main__":
+    root = tk.Tk()
+    app = FilterDevApp(root)
+    root.mainloop()

filter_dev/ble.py

@@ -0,0 +1,47 @@
+import matplotlib.pyplot as plt
+from time import time
+import asyncio
+from bleak import BleakClient, BleakScanner
+import pandas as pd
+from argparse import ArgumentParser
+from tqdm.auto import tqdm
+
+async def read_ble(reading_time=READING_TIME):
+    device = await BleakScanner.find_device_by_name("Smaage")
+    timestamps = []
+    raw_weights = []
+    
+    async with BleakClient(device.address) as client:
+        print(f"Connected to {device}")
+
+        # Read the characteristic value
+        pbar = tqdm(desc="reading data", total=reading_time)
+        time_start = time()
+        time_passed = 0
+        last_time_passed = 0
+        while time_passed < reading_time:
+            try:
+                # Read the sensor data
+                millis = await client.read_gatt_char(MILLIS_UUID)
+                millis = int.from_bytes(millis, byteorder='little')  # Adjust based on your data format
+                weight = await client.read_gatt_char(WEIGHT_UUID)
+                weight = int.from_bytes(weight, byteorder='little')  # Adjust based on your data format
+
+                timestamps.append(millis)
+                raw_weights.append(weight)
+
+            except Exception as e:
+                print(f"Error reading data: {e}")
+                break
+        
+            time_passed = time() - time_start
+            time_delta = (time_passed - last_time_passed)  
+            last_time_passed = time_passed
+            pbar.update(time_delta)
+        
+        pbar.close()
+        
+        df = pd.DataFrame({"timestamps": timestamps, "weights": raw_weights})
+        print(df['weights'].describe())
+        plt.plot(timestamps, raw_weights)
+        plt.show()

filter_dev/config.py

@@ -0,0 +1,4 @@
+# Replace with your Arduino's service and characteristic UUIDs
+SERVICE_UUID = "9f0dfdb2-e978-494c-8f15-68dbe8d28672"
+MILLIS_UUID = "abb92561-a809-453c-8c7c-71d3fff5b86e"
+WEIGHT_UUID = "123e4567-e89b-12d3-a456-426614174000"

filter_dev/filter/__init__.py

@@ -0,0 +1 @@
+from .mov_avg import MovAvg

filter_dev/filter/base.py

@@ -0,0 +1,36 @@
+from typing import Dict
+
+import pandas as pd
+from tkinter.ttk import Frame
+
+from ..gui.device import Device
+from ..gui.slider import Slider
+
+class Filter:
+
+    param_map: Dict[str, Slider] = {}
+
+    def __init__(self, device: Device, toolbar: Frame, callback: callable):
+        self.device = device
+        self.toolbar = toolbar
+        self.callback = callback
+
+        self.init_params(toolbar)
+
+    def init_params(self, toolbar):
+        raise NotImplementedError()
+
+
+    def _get_params(self):
+        params = {}
+        for k, v in self.param_map.items():
+            params[k] = v.get_value()
+        return params
+    
+    def __call__(self) -> pd.DataFrame:
+        df = self.device.data
+        df['filtered'] = self.filter(df)
+        return df
+    
+    def filter(self, df: pd.DataFrame) -> pd.Series:
+        raise NotImplementedError()

filter_dev/filter/mov_avg.py

@@ -0,0 +1,18 @@
+import pandas as pd
+
+from .base import Filter
+from ..gui.slider import Slider
+
+class MovAvg(Filter):
+    
+    def init_params(self, toolbar):
+        self.param_map = {
+            "window_size": Slider(toolbar, "Window Size", 1, 500, 10, self.callback),
+            "decimals": Slider(toolbar, "Decimals", 1, 5, 1, self.callback),
+            # "reset_threshold": Slider(self.toolbar, "Reset Threshold", 0.001, 0.1, 0.1, self.update),
+        }
+    
+    def filter(self, df: pd.DataFrame) -> pd.Series:
+        params = self._get_params()
+        return df['weights'].rolling(window=int(params['window_size'])).mean()\
+                            .round(int(params['decimals']))

filter_dev/gui/device.py

@@ -0,0 +1,40 @@
+from bleak import BleakClient, BleakScanner
+import pandas as pd
+from tkinter.ttk import Entry
+
+from ..config import MILLIS_UUID, WEIGHT_UUID
+
+class Device:
+
+    @property
+    def is_connected(self):
+        return self.device is not None
+    
+    @property
+    def data(self):
+        return pd.DataFrame({
+            "timestamps": self.timestamps,
+            "weights": self.weights
+        })
+
+    def __init__(self, device_name: Entry):
+        self.device = None
+        self.device_name = device_name
+        self.timestamps = []
+        self.weights = []
+    
+    async def connect(self):
+        self.device = await BleakScanner.find_device_by_name(self.device_name.get())
+
+        assert self.device is not None, "No Device found!"
+    
+    async def read_values(self):
+        assert self.is_connected, "Not connected"
+        async with BleakClient(self.device.address) as client:
+            millis = await client.read_gatt_char(MILLIS_UUID)
+            millis = int.from_bytes(millis, byteorder='little')  # Adjust based on your data format
+            weight = await client.read_gatt_char(WEIGHT_UUID)
+            weight = int.from_bytes(weight, byteorder='little')  # Adjust based on your data format
+
+            self.timestamps.append(millis)
+            self.weights.append(weight)

filter_dev/gui/slider.py

@@ -0,0 +1,28 @@
+from tkinter import ttk
+
+class Slider:
+    def __init__(self,
+                 parent,
+                 label_text,
+                 from_, to,
+                 initial_value,
+                 command):
+        self.command = command
+
+        self.frame = ttk.Frame(parent)
+        self.frame.pack(pady=10)
+
+        self.label = ttk.Label(self.frame, text=f"{label_text}: {int(initial_value)}")
+        self.label.pack()
+
+        self.slider = ttk.Scale(self.frame, from_=from_, to=to, orient='horizontal', command=self.update)
+        self.slider.set(initial_value)
+        self.slider.pack()
+
+    def update(self, event=None):
+        value = self.slider.get()
+        self.label.config(text=f"{self.label.cget('text').split(':')[0]}: {int(value)}")
+        self.command()
+
+    def get_value(self):
+        return self.slider.get()