# coding=utf-8 # # output_sparkfun_scmd_qwiic.py - Output for SparkFun Serial Controlled Motor Driver (SCMD) Qwiic # import threading import time import copy from flask_babel import lazy_gettext from mycodo.config_translations import TRANSLATIONS from mycodo.databases.models import OutputChannel from mycodo.outputs.base_output import AbstractOutput from mycodo.utils.constraints_pass import constraints_pass_percent from mycodo.utils.constraints_pass import constraints_pass_positive_value from mycodo.utils.database import db_retrieve_table_daemon from mycodo.utils.influx import add_measurements_influxdb def constraints_pass_motor_number(mod_input, value): """ Check if the motor number is valid (0-33) :param mod_input: SQL object with user-saved Input options :param value: integer :return: tuple: (bool, list of strings) """ errors = [] all_passed = True if value < 0 or value > 33: all_passed = False errors.append("Invalid motor number. Must be between 0 and 33") return all_passed, errors, mod_input # Measurements measurements_dict = { 0: { 'measurement': 'duration_time', 'unit': 's', 'name': 'Motor A On', }, 1: { 'measurement': 'volume', 'unit': 'ml', 'name': 'Motor A Dispense Volume', }, 2: { 'measurement': 'duration_time', 'unit': 's', 'name': 'Motor A Dispense Duration', }, 3: { 'measurement': 'duration_time', 'unit': 's', 'name': 'Motor B On', }, 4: { 'measurement': 'volume', 'unit': 'ml', 'name': 'Motor B Dispense Volume', }, 5: { 'measurement': 'duration_time', 'unit': 's', 'name': 'Motor B Dispense Duration', } } channels_dict = { 0: { 'name': 'Motor A', 'types': ['volume', 'on_off', 'pwm'], 'measurements': [0, 1, 2] }, 1: { 'name': 'Motor B', 'types': ['volume', 'on_off', 'pwm'], 'measurements': [3, 4, 5] } } # Output information OUTPUT_INFORMATION = { 'output_name_unique': 'sparkfun_scmd_qwiic', 'output_name': "SparkFun Serial Controlled Motor Driver (SCMD) Qwiic", 'output_manufacturer': 'SparkFun', 'output_library': 'sparkfun-qwiic-scmd', 'measurements_dict': measurements_dict, 'channels_dict': channels_dict, 'output_types': ['volume', 'on_off', 'pwm'], 'url_manufacturer': 'https://www.sparkfun.com/products/15451', 'url_datasheet': 'https://learn.sparkfun.com/tutorials/hookup-guide-for-the-qwiic-motor-driver', 'url_product_purchase': 'https://www.sparkfun.com/products/15451', 'message': 'Controls the SparkFun Serial Controlled Motor Driver (SCMD) via Qwiic I2C interface. ' 'Supports up to 34 motors (with expansion boards). Each channel can control one motor with ' 'variable speed (0-255) and direction (forward/reverse). Can be used for peristaltic pumps, ' 'DC motors, or other motor-driven devices.', 'options_enabled': [ 'i2c_location', 'button_on', 'button_send_volume', 'button_send_duration' ], 'options_disabled': ['interface'], 'dependencies_module': [ ('pip-pypi', 'qwiic-scmd', 'sparkfun-qwiic-scmd') ], 'interfaces': ['I2C'], 'i2c_address_editable': True, 'i2c_address_default': '0x5D', 'custom_options_message': "The SparkFun SCMD supports up to 34 motors when using expansion boards. " "For peristaltic pump applications, calibrate the flow rate by running the pump " "for 60 seconds at full speed and measuring the dispensed volume. Enter this value " "in the 'Fastest Rate (ml/min)' field for accurate volume dispensing.", 'custom_channel_options': [ { 'id': 'name', 'type': 'text', 'default_value': '', 'required': False, 'name': TRANSLATIONS['name']['title'], 'phrase': TRANSLATIONS['name']['phrase'] }, { 'id': 'motor_number', 'type': 'integer', 'default_value': 0, 'constraints_pass': constraints_pass_motor_number, 'name': 'Motor Number (0-33)', 'phrase': 'The motor port number on the SCMD board (0-33 with expansion boards)' }, { 'id': 'motor_speed', 'type': 'integer', 'default_value': 100, 'constraints_pass': constraints_pass_percent, 'name': 'Motor Speed (0-100%)', 'phrase': 'The motor output speed as a percentage (0-100)' }, { 'id': 'invert_direction', 'type': 'bool', 'default_value': False, 'name': 'Invert Direction', 'phrase': 'Invert the motor direction (swap forward/reverse)' }, { 'id': 'flow_mode', 'type': 'select', 'default_value': 'fastest_flow_rate', 'options_select': [ ('fastest_flow_rate', 'Fastest Flow Rate'), ('specify_flow_rate', 'Specify Flow Rate') ], 'name': lazy_gettext('Flow Rate Method'), 'phrase': 'The flow rate method to use when pumping a volume (for peristaltic pumps)' }, { 'id': 'flow_rate', 'type': 'float', 'default_value': 10.0, 'constraints_pass': constraints_pass_positive_value, 'name': 'Desired Flow Rate (ml/min)', 'phrase': 'Desired flow rate in ml/minute when Specify Flow Rate is set' }, { 'id': 'fastest_dispense_rate_ml_min', 'type': 'float', 'default_value': 100.0, 'constraints_pass': constraints_pass_positive_value, 'name': 'Fastest Rate (ml/min)', 'phrase': 'The fastest rate that the pump can dispense (ml/min) at 100% speed' }, { 'id': 'minimum_sec_on_per_min', 'type': 'float', 'default_value': 1.0, 'constraints_pass': constraints_pass_positive_value, 'name': 'Minimum On Time (sec/min)', 'phrase': 'Minimum seconds the pump should be on per minute for pulsing mode' } ], } class OutputModule(AbstractOutput): """An output support class that operates the SparkFun SCMD.""" def __init__(self, output, testing=False): super().__init__(output, testing=testing, name=__name__) self.scmd = None self.currently_dispensing = [False, False] # Track dispensing state for each channel output_channels = db_retrieve_table_daemon( OutputChannel).filter(OutputChannel.output_id == self.output.unique_id).all() self.options_channels = self.setup_custom_channel_options_json( OUTPUT_INFORMATION['custom_channel_options'], output_channels) def initialize(self): """Initialize the SCMD driver.""" from qwiic_scmd import QwiicScmd self.setup_on_off_output(OUTPUT_INFORMATION) try: self.logger.info("Initializing SparkFun SCMD on I2C bus {} at address {}".format( self.output.i2c_bus, self.output.i2c_location)) # Create SCMD instance self.scmd = QwiicScmd(address=int(str(self.output.i2c_location), 16)) # Check if device is connected if not self.scmd.is_connected(): self.logger.error("SCMD device not found at address {}".format( self.output.i2c_location)) return # Initialize the device if not self.scmd.begin(): self.logger.error("Failed to initialize SCMD device") return # Enable the driver self.scmd.enable() # Configure inversion mode for each channel if needed for channel in [0, 1]: if channel in self.options_channels['motor_number']: motor_num = self.options_channels['motor_number'][channel] invert = self.options_channels['invert_direction'][channel] if invert: self.scmd.inversion_mode(motor_num, 1) self.logger.info("Motor {} direction inverted".format(motor_num)) self.logger.info("SparkFun SCMD initialized successfully") self.output_setup = True except Exception as e: self.logger.exception("Could not set up SCMD output: {}".format(e)) return def output_switch(self, state, output_type=None, amount=None, duty_cycle=None, output_channel=None): """Control the motor output.""" if not self.is_setup(): msg = "Error 101: Device not set up. See https://kizniche.github.io/Mycodo/Error-Codes#error-101 for more info." self.logger.error(msg) return msg if output_channel is None: self.logger.error("output_channel is required") return motor_num = self.options_channels['motor_number'][output_channel] # Determine direction based on amount direction = 0 # 0 = forward, 1 = reverse if amount and amount < 0: direction = 1 amount = abs(amount) self.logger.debug( "Channel {}: state={}, type={}, amount={}, duty_cycle={}, motor={}, direction={}".format( output_channel, state, output_type, amount, duty_cycle, motor_num, "reverse" if direction else "forward")) try: # Turn off if state == 'off': if self.currently_dispensing[output_channel]: self.currently_dispensing[output_channel] = False self.logger.debug("Motor {} turned off".format(motor_num)) self.scmd.set_drive(motor_num, 0, 0) self.output_states[output_channel] = False # Volume dispensing (for peristaltic pumps) elif state == 'on' and output_type in ['vol', None] and amount not in [0, None]: if self.currently_dispensing[output_channel]: self.logger.debug( "Motor {} instructed to dispense while already dispensing. " "Overriding current operation.".format(motor_num)) if self.options_channels['flow_mode'][output_channel] == 'fastest_flow_rate': # Calculate time needed to dispense volume at fastest rate total_dispense_seconds = abs(amount) / self.options_channels['fastest_dispense_rate_ml_min'][output_channel] * 60 msg = "Motor {}: Dispensing {:.1f} ml in {:.1f} seconds at fastest rate ({:.1f} ml/min)".format( motor_num, abs(amount), total_dispense_seconds, self.options_channels['fastest_dispense_rate_ml_min'][output_channel]) self.logger.info(msg) write_db = threading.Thread( target=self.dispense_volume_fastest, args=(output_channel, motor_num, direction, amount, total_dispense_seconds)) write_db.start() return elif self.options_channels['flow_mode'][output_channel] == 'specify_flow_rate': # Calculate time and duty cycle for specified flow rate write_db = threading.Thread( target=self.dispense_volume_rate, args=(output_channel, motor_num, direction, amount)) write_db.start() return # Duration-based operation elif state == 'on' and output_type == 'sec' and amount: if self.currently_dispensing[output_channel]: self.logger.debug( "Motor {} instructed to turn on while already running. " "Overriding current operation.".format(motor_num)) speed_percent = self.options_channels['motor_speed'][output_channel] speed_level = int(speed_percent * 255 / 100) self.logger.info("Motor {}: Running for {:.1f} seconds at {}% speed".format( motor_num, amount, speed_percent)) self.scmd.set_drive(motor_num, direction, speed_level) self.output_states[output_channel] = True # Schedule turn off write_db = threading.Thread( target=self.run_for_duration, args=(output_channel, motor_num, amount)) write_db.start() return # PWM / Direct speed control elif state == 'on' and (output_type == 'pwm' or duty_cycle is not None): if duty_cycle is not None: speed_level = int(duty_cycle * 255 / 100) else: speed_percent = self.options_channels['motor_speed'][output_channel] speed_level = int(speed_percent * 255 / 100) self.logger.info("Motor {}: Setting speed to {} ({}%)".format( motor_num, speed_level, speed_level * 100 / 255)) self.scmd.set_drive(motor_num, direction, speed_level) self.output_states[output_channel] = True # Simple on/off elif state == 'on': speed_percent = self.options_channels['motor_speed'][output_channel] speed_level = int(speed_percent * 255 / 100) self.logger.info("Motor {}: Turned on at {}% speed".format(motor_num, speed_percent)) self.scmd.set_drive(motor_num, direction, speed_level) self.output_states[output_channel] = True else: self.logger.error( "Invalid parameters: state={}, type={}, amount={}, duty_cycle={}".format( state, output_type, amount, duty_cycle)) return except Exception as e: self.logger.exception("Error controlling motor {}: {}".format(motor_num, e)) return str(e) def dispense_volume_fastest(self, channel, motor_num, direction, amount, total_dispense_seconds): """Dispense volume at fastest flow rate (100% duty cycle).""" self.currently_dispensing[channel] = True self.logger.debug("Motor {}: Starting fastest dispense".format(motor_num)) try: # Turn on at full speed self.scmd.set_drive(motor_num, direction, 255) self.output_states[channel] = True timer_dispense = time.time() + total_dispense_seconds while time.time() < timer_dispense and self.currently_dispensing[channel]: time.sleep(0.01) # Turn off self.scmd.set_drive(motor_num, 0, 0) self.output_states[channel] = False self.currently_dispensing[channel] = False self.logger.info("Motor {}: Dispense complete".format(motor_num)) self.record_dispersal(channel, amount, total_dispense_seconds, total_dispense_seconds) except Exception as e: self.logger.exception("Error during dispense: {}".format(e)) self.currently_dispensing[channel] = False def dispense_volume_rate(self, channel, motor_num, direction, amount): """Dispense volume at specified flow rate using PWM.""" self.currently_dispensing[channel] = True try: fastest_rate = self.options_channels['fastest_dispense_rate_ml_min'][channel] desired_rate = self.options_channels['flow_rate'][channel] min_sec_on = self.options_channels['minimum_sec_on_per_min'][channel] # Calculate slowest possible rate slowest_rate = fastest_rate / 60 * min_sec_on # Clamp desired rate to valid range if desired_rate < slowest_rate: self.logger.warning("Desired rate {:.1f} ml/min too slow, using {:.1f} ml/min".format( desired_rate, slowest_rate)) desired_rate = slowest_rate elif desired_rate > fastest_rate: self.logger.warning("Desired rate {:.1f} ml/min too fast, using {:.1f} ml/min".format( desired_rate, fastest_rate)) desired_rate = fastest_rate # Calculate duty cycle and timing duty_cycle = desired_rate / fastest_rate total_dispense_seconds = abs(amount) / desired_rate * 60 self.logger.info("Motor {}: Dispensing {:.1f} ml at {:.1f} ml/min (duty cycle: {:.1%})".format( motor_num, abs(amount), desired_rate, duty_cycle)) # Use PWM by pulsing the motor pulse_period = 1.0 # 1 second pulse period on_time = pulse_period * duty_cycle off_time = pulse_period - on_time timer_dispense = time.time() + total_dispense_seconds total_on_time = 0 while time.time() < timer_dispense and self.currently_dispensing[channel]: # Turn on self.scmd.set_drive(motor_num, direction, 255) self.output_states[channel] = True time_on_start = time.time() while (time.time() - time_on_start) < on_time and self.currently_dispensing[channel]: time.sleep(0.01) total_on_time += time.time() - time_on_start # Turn off self.scmd.set_drive(motor_num, 0, 0) self.output_states[channel] = False time_off_start = time.time() while (time.time() - time_off_start) < off_time and self.currently_dispensing[channel]: time.sleep(0.01) # Ensure motor is off self.scmd.set_drive(motor_num, 0, 0) self.output_states[channel] = False self.currently_dispensing[channel] = False self.logger.info("Motor {}: Dispense complete".format(motor_num)) self.record_dispersal(channel, amount, total_on_time, total_dispense_seconds) except Exception as e: self.logger.exception("Error during rate dispense: {}".format(e)) self.currently_dispensing[channel] = False def run_for_duration(self, channel, motor_num, duration): """Run motor for a specific duration then turn off.""" try: time.sleep(duration) if self.output_states[channel]: self.scmd.set_drive(motor_num, 0, 0) self.output_states[channel] = False self.logger.info("Motor {}: Duration complete, turned off".format(motor_num)) except Exception as e: self.logger.exception("Error in run_for_duration: {}".format(e)) def record_dispersal(self, channel, amount, total_on_seconds, total_dispense_seconds): """Record dispersal measurements to database.""" try: measure_dict = copy.deepcopy(measurements_dict) base_index = channel * 3 measure_dict[base_index]['value'] = total_on_seconds measure_dict[base_index + 1]['value'] = abs(amount) measure_dict[base_index + 2]['value'] = total_dispense_seconds add_measurements_influxdb(self.unique_id, measure_dict) except Exception as e: self.logger.exception("Error recording dispersal: {}".format(e)) def is_on(self, output_channel=None): """Check if output is currently on.""" if self.is_setup(): if output_channel is not None: return self.output_states.get(output_channel, False) return any(self.output_states.values()) return False def is_setup(self): """Check if output has been set up successfully.""" return self.output_setup def stop_output(self): """Stop all motors when output is disabled.""" if self.scmd: try: for channel in [0, 1]: if channel in self.options_channels['motor_number']: motor_num = self.options_channels['motor_number'][channel] self.scmd.set_drive(motor_num, 0, 0) self.currently_dispensing[channel] = False self.scmd.disable() self.logger.info("All motors stopped and SCMD disabled") except Exception as e: self.logger.exception("Error stopping motors: {}".format(e))