# coding=utf-8 import copy import threading import time 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, constraints_pass_positive_value from mycodo.utils.database import db_retrieve_table_daemon from mycodo.utils.influx import add_measurements_influxdb measurements_dict = { 0: {'measurement': 'duration_time', 'unit': 's', 'name': 'Pump On Motor A'}, 1: {'measurement': 'volume', 'unit': 'ml', 'name': 'Dispense Volume Motor A'}, 2: {'measurement': 'duration_time', 'unit': 's', 'name': 'Dispense Duration Motor A'}, 3: {'measurement': 'duration_time', 'unit': 's', 'name': 'Pump On Motor B'}, 4: {'measurement': 'volume', 'unit': 'ml', 'name': 'Dispense Volume Motor B'}, 5: {'measurement': 'duration_time', 'unit': 's', 'name': 'Dispense Duration Motor B'} } channels_dict = { 0: {'name': 'Motor A', 'types': ['volume', 'on_off'], 'measurements': [0, 1, 2]}, 1: {'name': 'Motor B', 'types': ['volume', 'on_off'], 'measurements': [3, 4, 5]} } OUTPUT_INFORMATION = { 'output_name_unique': 'qwiic_scmd', 'output_name': "{}: Sparkfun Qwiic SCMD Motor Driver".format(lazy_gettext('Peristaltic Pump')), 'output_manufacturer': 'Sparkfun', 'output_library': 'qwiic_scmd', 'measurements_dict': measurements_dict, 'channels_dict': channels_dict, 'output_types': ['volume', 'on_off'], 'url_manufacturer': 'https://learn.sparkfun.com/tutorials/hookup-guide-for-the-qwiic-serial-control-motor-driver', 'url_datasheet': 'https://cdn.sparkfun.com/assets/5/5/1/8/d/SparkFun_Serial_Controlled_Motor_Driver.pdf', 'message': 'Controls 2 DC motors; can dispense volumes accurately.', '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_channel_options': [ {'id': 'name', 'type': 'text', 'default_value': '', 'required': False, 'name': TRANSLATIONS['name']['title'], 'phrase': TRANSLATIONS['name']['phrase']}, {'id': 'motor_speed', 'type': 'integer', 'default_value': 100, 'constraints_pass': constraints_pass_percent, 'name': 'Motor Speed (0-100)', 'phrase': 'Motor speed in %'}, {'id': 'direction', 'type': 'select', 'default_value': 1, 'options_select': [(1, 'Forward'), (0, 'Reverse')], 'name': lazy_gettext('Direction'), 'phrase': 'Motor rotation direction'}, {'id': 'flow_rate', 'type': 'float', 'default_value': 10.0, 'constraints_pass': constraints_pass_positive_value, 'name': 'Flow Rate (ml/min)', 'phrase': 'Measured flow rate at set speed'} ] } class OutputModule(AbstractOutput): def __init__(self, output, testing=False): super().__init__(output, testing=testing, name=__name__) self.scmd = None self.output_setup = False self.currently_dispensing = {} self.OUTPUT_INFORMATION = OUTPUT_INFORMATION # Initialize all expected attributes self.output_states = {} self.output_time_turned_on = {} self.output_off_until = {} self.output_last_duration = {} self.output_pwm = {} self.output_off_triggered = {} for ch in OUTPUT_INFORMATION['channels_dict']: self.output_states[ch] = False self.output_time_turned_on[ch] = None self.output_off_until[ch] = None self.output_last_duration[ch] = None self.output_pwm[ch] = None self.output_off_triggered[ch] = False self.currently_dispensing[ch] = False # Load channel options 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): try: import qwiic_scmd except ImportError: self.logger.error("qwiic_scmd library not installed.") return False try: i2c_addr = int(str(self.output.i2c_location), 16) if self.output.i2c_location else 0x5D self.scmd = qwiic_scmd.QwiicScmd(address=i2c_addr) if not self.scmd.begin(): self.logger.error("SCMD begin() failed") return False self.scmd.enable() self.scmd.set_drive(0, 1, 0) self.scmd.set_drive(1, 1, 0) self.output_setup = True self.logger.info("SCMD initialized") return True except Exception as e: self.logger.exception(f"Error initializing SCMD: {e}") return False def output_switch(self, state, output_type=None, amount=None, output_channel=0): if not self.is_setup(): self.logger.error("Output not set up") return "Output not set up" motor_id = output_channel direction = int(self.options_channels['direction'].get(output_channel, 1)) motor_speed = int(self.options_channels['motor_speed'].get(output_channel, 100)) motor_power = int(motor_speed * 255 / 100) if state == 'on': if output_type == 'volume' and amount: if self.currently_dispensing[output_channel]: self.logger.warning("Already dispensing, overriding.") flow_rate = float(self.options_channels['flow_rate'].get(output_channel, 10.0)) total_sec = max(amount / flow_rate * 60, 0.1) t = threading.Thread(target=self._dispense_volume, args=(output_channel, motor_id, direction, amount, total_sec)) t.start() return "Success" else: self.scmd.set_drive(motor_id, direction, motor_power) self.output_states[output_channel] = True self.output_time_turned_on[output_channel] = time.time() return "Success" elif state == 'off': self.currently_dispensing[output_channel] = False self.scmd.set_drive(motor_id, 0, 0) self.output_states[output_channel] = False self.output_time_turned_on[output_channel] = None return "Success" else: self.logger.error("Invalid state") return "Invalid state" def _dispense_volume(self, ch, motor_id, direction, amount, seconds): self.currently_dispensing[ch] = True motor_speed = int(self.options_channels['motor_speed'].get(ch, 100)) motor_power = int(motor_speed * 255 / 100) self.scmd.set_drive(motor_id, direction, motor_power) self.output_states[ch] = True self.output_time_turned_on[ch] = time.time() end_time = time.time() + seconds while time.time() < end_time and self.currently_dispensing[ch]: time.sleep(0.01) self.scmd.set_drive(motor_id, 0, 0) self.output_states[ch] = False self.output_time_turned_on[ch] = None self.currently_dispensing[ch] = False measure = copy.deepcopy(measurements_dict) if ch == 0: measure[0]['value'] = seconds measure[1]['value'] = amount measure[2]['value'] = seconds else: measure[3]['value'] = seconds measure[4]['value'] = amount measure[5]['value'] = seconds add_measurements_influxdb(self.unique_id, measure) def is_setup(self): return self.output_setup and self.scmd is not None def is_on(self, output_channel=None): if output_channel is not None: return self.output_states.get(output_channel, False) or self.currently_dispensing.get(output_channel, False) else: return any(self.output_states.values()) or any(self.currently_dispensing.values()) def stop_output(self): if self.is_setup(): for ch in [0, 1]: self.scmd.set_drive(ch, 0, 0) self.output_states[ch] = False self.currently_dispensing[ch] = False self.output_time_turned_on[ch] = None try: self.scmd.disable() except Exception as e: self.logger.error(f"Error disabling SCMD: {e}") def output_sec_currently_on(self, output_channel): start = self.output_time_turned_on.get(output_channel) return time.time() - start if start else 0