From 79d441a86e0874440ab7a626a93a456921d53307 Mon Sep 17 00:00:00 2001 From: schn33fuchs Date: Sun, 7 Jun 2026 10:27:22 +0200 Subject: [PATCH] Motor The raspi now controls a motor to lock unlock --- raspi/motor.py | 110 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_motor.py | 40 ++++++++++++++++ ui/cli.py | 40 ++++++++++++---- ui/kivy.py | 34 ++++++++++++++ 4 files changed, 214 insertions(+), 10 deletions(-) create mode 100644 raspi/motor.py create mode 100644 tests/test_motor.py diff --git a/raspi/motor.py b/raspi/motor.py new file mode 100644 index 0000000..b2b4dfa --- /dev/null +++ b/raspi/motor.py @@ -0,0 +1,110 @@ +import sys +import time +import logging +import threading + +logger = logging.getLogger(__name__) + +try: + if "--no-raspi" in sys.argv or "--mock-motor" in sys.argv: + raise ImportError("RPi functionality disabled via command line flag.") + import RPi.GPIO as GPIO + GPIO_AVAILABLE = True +except (ImportError, RuntimeError): + GPIO_AVAILABLE = False + GPIO = None + + + + +class Raspi: + CLOCKWISE = False + COUNTER_CLOCKWISE = True + + def __init__(self) -> None: + self.in1 = 3 + self.in2 = 5 + self.in3 = 7 + self.in4 = 11 + + self.step_sleep = 0.002 + self.step_sequence = [ + [True, False, False, True], + [True, False, False, False], + [True, True, False, False], + [False, True, False, False], + [False, True, True, False], + [False, False, True, False], + [False, False, True, True], + [False, False, False, True], + ] + + self.motor_pins = [self.in1, self.in2, self.in3, self.in4] + self.motor_step_counter = 0 + self.initialized = False + + def _setup_gpio(self): + if not GPIO_AVAILABLE: + logger.info("GPIO not available, skipping GPIO setup.") + return + if not self.initialized: + try: + GPIO.setmode(GPIO.BCM) + for pin in self.motor_pins: + GPIO.setup(pin, GPIO.OUT) + GPIO.output(pin, GPIO.LOW) + self.initialized = True + except Exception as e: + logger.error(f"Failed to setup GPIO: {e}") + + def cleanup(self): + if not GPIO_AVAILABLE: + logger.info("GPIO not available, skipping GPIO cleanup.") + return + try: + for pin in self.motor_pins: + GPIO.output(pin, GPIO.LOW) + GPIO.cleanup() + self.initialized = False + except Exception as e: + logger.error(f"Failed to cleanup GPIO: {e}") + + def move(self, step_count, direction): + self._setup_gpio() + if not GPIO_AVAILABLE: + dir_str = ( + "clockwise" if direction == self.CLOCKWISE else "counter-clockwise" + ) + logger.info(f"[Mock Motor] Moving {step_count} steps {dir_str}.") + return + + try: + for _i in range(step_count): + for pin in range(0, len(self.motor_pins)): + GPIO.output( + self.motor_pins[pin], + self.step_sequence[self.motor_step_counter][pin], + ) + if direction == self.COUNTER_CLOCKWISE: + self.motor_step_counter = (self.motor_step_counter - 1) % 8 + else: + self.motor_step_counter = (self.motor_step_counter + 1) % 8 + time.sleep(self.step_sleep) + except Exception as e: + logger.error(f"Error during motor movement: {e}") + finally: + self.cleanup() + + def unlock(self): + # 90 degrees clockwise (1024 steps) + self.move(1024, self.CLOCKWISE) + + def lock(self): + # 90 degrees counter-clockwise (1024 steps) + self.move(1024, self.COUNTER_CLOCKWISE) + + def unlock_async(self): + threading.Thread(target=self.unlock, daemon=True).start() + + def lock_async(self): + threading.Thread(target=self.lock, daemon=True).start() diff --git a/tests/test_motor.py b/tests/test_motor.py new file mode 100644 index 0000000..d9e4f09 --- /dev/null +++ b/tests/test_motor.py @@ -0,0 +1,40 @@ +import unittest +from raspi.motor import Raspi + + +class TestMotor(unittest.TestCase): + def setUp(self): + self.motor = Raspi() + + def test_mock_unlock(self): + # Unlocking should not raise any errors, even if GPIO is not available + try: + self.motor.unlock() + success = True + except Exception: + success = False + self.assertTrue(success) + + def test_mock_lock(self): + try: + self.motor.lock() + success = True + except Exception: + success = False + self.assertTrue(success) + + def test_mock_async_unlock(self): + try: + self.motor.unlock_async() + success = True + except Exception: + success = False + self.assertTrue(success) + + def test_mock_async_lock(self): + try: + self.motor.lock_async() + success = True + except Exception: + success = False + self.assertTrue(success) diff --git a/ui/cli.py b/ui/cli.py index 9028290..7cfd6b9 100644 --- a/ui/cli.py +++ b/ui/cli.py @@ -5,6 +5,9 @@ class CLI: def __init__(self, tracker) -> None: self.tracker = tracker self.current_user = None + from raspi.motor import Raspi + self.motor = Raspi() + def run(self): while True: @@ -157,20 +160,27 @@ class CLI: if confirm == "y": result = self.tracker.add_purchase(self.current_user, item_id) - self._clear_screen() - print("=" * 50) - print("PURCHASE ADDED") - print("=" * 50) - print() - print(f"✓ {result['message']}") - print(f"New debt: ${result['new_debt']:.2f}") - print() + if result["status"] == "success": + self._clear_screen() + print("=" * 50) + print("PURCHASE ADDED") + print("=" * 50) + print() + print(f"✓ {result['message']}") + print(f"New debt: ${result['new_debt']:.2f}") + print() - input("Press Enter to continue...") + self.motor.unlock_async() + input("Press Enter to continue...") + self.motor.lock_async() + else: + print(f"Error: {result['message']}") + input("Press Enter to continue...") else: print("Purchase cancelled.") input("Press Enter to continue...") + def _view_user_debt(self): self._clear_screen() @@ -237,7 +247,9 @@ class CLI: print("4. View transaction history") print("5. Manage users") print("6. Manage items") - print("7. Logout") + print("7. Open Kühlschrank (Unlock)") + print("8. Close Kühlschrank (Lock)") + print("9. Logout") print() choice = input("Choose option: ").strip() @@ -255,6 +267,14 @@ class CLI: elif choice == "6": self._manage_items() elif choice == "7": + print("Kühlschrank wird geöffnet (Entriegeln)...") + self.motor.unlock_async() + input("Press Enter to continue...") + elif choice == "8": + print("Kühlschrank wird geschlossen (Verriegeln)...") + self.motor.lock_async() + input("Press Enter to continue...") + elif choice == "9": break else: self._invalid_choice() diff --git a/ui/kivy.py b/ui/kivy.py index 724676c..97a4b3e 100644 --- a/ui/kivy.py +++ b/ui/kivy.py @@ -1305,6 +1305,8 @@ class AddPurchaseScreen(StyledScreen): title="Erfolg", message=f"Kauf erfolgreich!\n\nNeuer Schuldenstand: {result['new_debt']:.2f} €", ) + popup.bind(on_dismiss=lambda instance: app.motor.lock_async()) + app.motor.unlock_async() popup.open() self.manager.current = "user_menu" else: @@ -1312,6 +1314,7 @@ class AddPurchaseScreen(StyledScreen): popup.open() + class AllDebtsScreen(StyledScreen): def __init__(self, **kwargs): super(AllDebtsScreen, self).__init__( @@ -1531,6 +1534,26 @@ class AdminMenuScreen(StyledScreen): btn_manage_items.bind(on_release=self.go_manage_items) box.add_widget(btn_manage_items) + btn_open = RoundedButton( + text="Kühlschrank öffnen", + size_hint=(0.7, None), + height="60dp", + pos_hint={"center_x": 0.5}, + btn_color=[0.0, 0.65, 0.57, 1], + ) + btn_open.bind(on_release=self.open_fridge) + box.add_widget(btn_open) + + btn_close = RoundedButton( + text="Kühlschrank schließen", + size_hint=(0.7, None), + height="60dp", + pos_hint={"center_x": 0.5}, + btn_color=[0.8, 0.2, 0.2, 1], + ) + btn_close.bind(on_release=self.close_fridge) + box.add_widget(btn_close) + btn_logout = RoundedButton( text="Admin-Bereich verlassen", size_hint=(0.7, None), @@ -1541,6 +1564,14 @@ class AdminMenuScreen(StyledScreen): btn_logout.bind(on_release=self.logout) box.add_widget(btn_logout) + def open_fridge(self, instance): + app = App.get_running_app() + app.motor.unlock_async() + + def close_fridge(self, instance): + app = App.get_running_app() + app.motor.lock_async() + def go_all_debts(self, instance): self.manager.get_screen("all_debts").back_screen = "admin_menu" self.manager.current = "all_debts" @@ -1994,6 +2025,9 @@ class KuehlschrankApp(App): self.tracker = tracker self.current_user = None self.admin_password = None + from raspi.motor import Raspi + self.motor = Raspi() + def build(self): self.title = "Getränkeliste"