jensandClaude Sonnet 5
f20e617d81
Heater and Stirrer can be real serial hardware (hendi, Pololu1376), but there was no connection concept at all - the constructors opened the port and crashed the whole server if the device was missing, with no way to see connection status or firmware version and no way to reconnect without a restart. Adds an observable Connectable mixin (components/connectable.py) shared by AHeater/AStirrer; real devices defer opening the serial port to an explicit connect(), auto-connect on server startup, and surface Connected/FirmwareVersion/Simulated plus manual Connect/Disconnect over the web GUI. Heating/stirring and Sud Start are all gated on connection state, and a disconnect mid-brew force-stops the run via the same path as a manual Stop. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YaPLuRPpyjWcwhMvCvpHCL
116 lines
2.5 KiB
Python
116 lines
2.5 KiB
Python
from contextlib import contextmanager
|
|
from components import AStirrer
|
|
from components.actor.pololu1376 import *
|
|
import time
|
|
import serial
|
|
|
|
|
|
class StirrerPololu1376(AStirrer):
|
|
def name(self):
|
|
return "Pololu1376"
|
|
|
|
def __init__(self, dt, params):
|
|
AStirrer.__init__(self, dt, simulated=False)
|
|
|
|
# Deferred-open pyserial idiom - see Pololu1376's own constructor
|
|
# comment: connect() is what actually opens params['port'] (and
|
|
# sets output flow control, which itself requires an open port).
|
|
ser = serial.Serial()
|
|
ser.port = params['port']
|
|
ser.baudrate = params['speed']
|
|
self.drv = Pololu1376(ser)
|
|
|
|
def connect(self):
|
|
ok = self.drv.connect()
|
|
self.connected = ok
|
|
self.firmware_version = self.drv.firmware_version if ok else None
|
|
return ok
|
|
|
|
def disconnect(self):
|
|
self.drv.disconnect()
|
|
self.connected = False
|
|
self.firmware_version = None
|
|
|
|
@contextmanager
|
|
def remote_open(self):
|
|
try:
|
|
self.activate(True)
|
|
yield None
|
|
finally:
|
|
self.activate(False)
|
|
|
|
def get_speed(self):
|
|
if self.isOn and self.is_activated():
|
|
return self.speed
|
|
return 0
|
|
|
|
def activate(self, enable):
|
|
if not self.connected:
|
|
return
|
|
print("activate {}".format(enable))
|
|
if enable:
|
|
self.drv.go()
|
|
else:
|
|
self.drv.stop()
|
|
|
|
def _on_set_speed(self, speed):
|
|
if not self.connected:
|
|
return
|
|
self.drv.motor_forward(speed)
|
|
print("Set speed to {} %".format(speed))
|
|
|
|
def _on_process(self):
|
|
if not self.connected:
|
|
return
|
|
status = self.drv.get_variable(Varid.STATUS)
|
|
limit_status = self.drv.get_variable(Varid.STATUS_LIMIT_STATUS)
|
|
if status & 0x01 == 0x01:
|
|
if limit_status & 0x01 == 0x01:
|
|
print("Recover after motor error!")
|
|
self.drv.go()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
s = StirrerPololu1376(dt=1.0, params={'port': "/dev/ttyACM0", 'speed': 115200})
|
|
s.connect()
|
|
|
|
with s.remote_open():
|
|
|
|
print("Set some speeds")
|
|
s.set_speed(50)
|
|
time.sleep(2)
|
|
s.set_speed(20)
|
|
time.sleep(2)
|
|
s.set_speed(50)
|
|
time.sleep(2)
|
|
|
|
print("Pulsed operation, duty cycle = 50%")
|
|
s.set_duty_cycle(0.5)
|
|
s.set_cycle_time(10)
|
|
for i in range(1, 30):
|
|
s.process()
|
|
time.sleep(1)
|
|
|
|
print("Pulsed operation, duty cycle = 20%")
|
|
s.set_duty_cycle(0.2)
|
|
s.set_cycle_time(10)
|
|
for i in range(1, 30):
|
|
s.process()
|
|
time.sleep(1)
|
|
|
|
print("Continious operation")
|
|
s.set_duty_cycle(1.0)
|
|
s.set_speed(50)
|
|
time.sleep(3.0)
|
|
|
|
print("Continious operation increasing speed")
|
|
for speed in range (25, 75, 5):
|
|
s.set_duty_cycle(1.0)
|
|
s.set_speed(speed)
|
|
for d in range(0,3,1):
|
|
s.process()
|
|
time.sleep(1.0)
|
|
|
|
print("Leaving context")
|
|
|