111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
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()
|