diff options
Diffstat (limited to 'klippy/extras/bus.py')
| -rw-r--r-- | klippy/extras/bus.py | 245 |
1 files changed, 154 insertions, 91 deletions
diff --git a/klippy/extras/bus.py b/klippy/extras/bus.py index 4121c1c8..6e87b2e1 100644 --- a/klippy/extras/bus.py +++ b/klippy/extras/bus.py @@ -5,10 +5,11 @@ # This file may be distributed under the terms of the GNU GPLv3 license. import mcu + def resolve_bus_name(mcu, param, bus): # Find enumerations for the given bus enumerations = mcu.get_enumerations() - enums = enumerations.get(param, enumerations.get('bus')) + enums = enumerations.get(param, enumerations.get("bus")) if enums is None: if bus is None: return 0 @@ -25,10 +26,10 @@ def resolve_bus_name(mcu, param, bus): raise ppins.error("Unknown %s '%s'" % (param, bus)) # Check for reserved bus pins constants = mcu.get_constants() - reserve_pins = constants.get('BUS_PINS_%s' % (bus,), None) + reserve_pins = constants.get("BUS_PINS_%s" % (bus,), None) pin_resolver = ppins.get_pin_resolver(mcu_name) if reserve_pins is not None: - for pin in reserve_pins.split(','): + for pin in reserve_pins.split(","): pin_resolver.reserve_pin(pin, bus) return bus @@ -37,10 +38,10 @@ def resolve_bus_name(mcu, param, bus): # SPI ###################################################################### + # Helper code for working with devices connected to an MCU via an SPI bus class MCU_SPI: - def __init__(self, mcu, bus, pin, mode, speed, sw_pins=None, - cs_active_high=False): + def __init__(self, mcu, bus, pin, mode, speed, sw_pins=None, cs_active_high=False): self.mcu = mcu self.bus = bus self.speed = speed @@ -49,101 +50,133 @@ class MCU_SPI: if pin is None: mcu.add_config_cmd("config_spi_without_cs oid=%d" % (self.oid,)) else: - mcu.add_config_cmd("config_spi oid=%d pin=%s cs_active_high=%d" - % (self.oid, pin, cs_active_high)) + mcu.add_config_cmd( + "config_spi oid=%d pin=%s cs_active_high=%d" + % (self.oid, pin, cs_active_high) + ) # Generate SPI bus config message self.config_fmt_ticks = None if sw_pins is not None: self.config_fmt = ( "spi_set_software_bus oid=%d" " miso_pin=%s mosi_pin=%s sclk_pin=%s mode=%d rate=%d" - % (self.oid, sw_pins[0], sw_pins[1], sw_pins[2], mode, speed)) + % (self.oid, sw_pins[0], sw_pins[1], sw_pins[2], mode, speed) + ) self.config_fmt_ticks = ( "spi_set_sw_bus oid=%d" " miso_pin=%s mosi_pin=%s sclk_pin=%s mode=%d pulse_ticks=%%d" - % (self.oid, sw_pins[0], sw_pins[1], - sw_pins[2], mode)) + % (self.oid, sw_pins[0], sw_pins[1], sw_pins[2], mode) + ) else: - self.config_fmt = ( - "spi_set_bus oid=%d spi_bus=%%s mode=%d rate=%d" - % (self.oid, mode, speed)) + self.config_fmt = "spi_set_bus oid=%d spi_bus=%%s mode=%d rate=%d" % ( + self.oid, + mode, + speed, + ) self.cmd_queue = mcu.alloc_command_queue() mcu.register_config_callback(self.build_config) self.spi_send_cmd = self.spi_transfer_cmd = None + def setup_shutdown_msg(self, shutdown_seq): shutdown_msg = "".join(["%02x" % (x,) for x in shutdown_seq]) self.mcu.add_config_cmd( "config_spi_shutdown oid=%d spi_oid=%d shutdown_msg=%s" - % (self.mcu.create_oid(), self.oid, shutdown_msg)) + % (self.mcu.create_oid(), self.oid, shutdown_msg) + ) + def get_oid(self): return self.oid + def get_mcu(self): return self.mcu + def get_command_queue(self): return self.cmd_queue + def build_config(self): - if '%' in self.config_fmt: + if "%" in self.config_fmt: bus = resolve_bus_name(self.mcu, "spi_bus", self.bus) self.config_fmt = self.config_fmt % (bus,) if self.config_fmt_ticks: - if self.mcu.try_lookup_command("spi_set_sw_bus oid=%c miso_pin=%u " - "mosi_pin=%u sclk_pin=%u " - "mode=%u pulse_ticks=%u"): - pulse_ticks = self.mcu.seconds_to_clock(1./self.speed) + if self.mcu.try_lookup_command( + "spi_set_sw_bus oid=%c miso_pin=%u " + "mosi_pin=%u sclk_pin=%u " + "mode=%u pulse_ticks=%u" + ): + pulse_ticks = self.mcu.seconds_to_clock(1.0 / self.speed) self.config_fmt = self.config_fmt_ticks % (pulse_ticks,) self.mcu.add_config_cmd(self.config_fmt) self.spi_send_cmd = self.mcu.lookup_command( - "spi_send oid=%c data=%*s", cq=self.cmd_queue) + "spi_send oid=%c data=%*s", cq=self.cmd_queue + ) self.spi_transfer_cmd = self.mcu.lookup_query_command( "spi_transfer oid=%c data=%*s", - "spi_transfer_response oid=%c response=%*s", oid=self.oid, - cq=self.cmd_queue) + "spi_transfer_response oid=%c response=%*s", + oid=self.oid, + cq=self.cmd_queue, + ) + def spi_send(self, data, minclock=0, reqclock=0): if self.spi_send_cmd is None: # Send setup message via mcu initialization data_msg = "".join(["%02x" % (x,) for x in data]) - self.mcu.add_config_cmd("spi_send oid=%d data=%s" % ( - self.oid, data_msg), is_init=True) + self.mcu.add_config_cmd( + "spi_send oid=%d data=%s" % (self.oid, data_msg), is_init=True + ) return - self.spi_send_cmd.send([self.oid, data], - minclock=minclock, reqclock=reqclock) + self.spi_send_cmd.send([self.oid, data], minclock=minclock, reqclock=reqclock) + def spi_transfer(self, data, minclock=0, reqclock=0): - return self.spi_transfer_cmd.send([self.oid, data], - minclock=minclock, reqclock=reqclock) - def spi_transfer_with_preface(self, preface_data, data, - minclock=0, reqclock=0): + return self.spi_transfer_cmd.send( + [self.oid, data], minclock=minclock, reqclock=reqclock + ) + + def spi_transfer_with_preface(self, preface_data, data, minclock=0, reqclock=0): return self.spi_transfer_cmd.send_with_preface( - self.spi_send_cmd, [self.oid, preface_data], [self.oid, data], - minclock=minclock, reqclock=reqclock) + self.spi_send_cmd, + [self.oid, preface_data], + [self.oid, data], + minclock=minclock, + reqclock=reqclock, + ) + # Helper to setup an spi bus from settings in a config section -def MCU_SPI_from_config(config, mode, pin_option="cs_pin", - default_speed=100000, share_type=None, - cs_active_high=False): +def MCU_SPI_from_config( + config, + mode, + pin_option="cs_pin", + default_speed=100000, + share_type=None, + cs_active_high=False, +): # Determine pin from config ppins = config.get_printer().lookup_object("pins") cs_pin = config.get(pin_option) cs_pin_params = ppins.lookup_pin(cs_pin, share_type=share_type) - pin = cs_pin_params['pin'] - if pin == 'None': + pin = cs_pin_params["pin"] + if pin == "None": ppins.reset_pin_sharing(cs_pin_params) pin = None # Load bus parameters - mcu = cs_pin_params['chip'] - speed = config.getint('spi_speed', default_speed, minval=100000) - if config.get('spi_software_sclk_pin', None) is not None: - sw_pin_names = ['spi_software_%s_pin' % (name,) - for name in ['miso', 'mosi', 'sclk']] - sw_pin_params = [ppins.lookup_pin(config.get(name), share_type=name) - for name in sw_pin_names] + mcu = cs_pin_params["chip"] + speed = config.getint("spi_speed", default_speed, minval=100000) + if config.get("spi_software_sclk_pin", None) is not None: + sw_pin_names = [ + "spi_software_%s_pin" % (name,) for name in ["miso", "mosi", "sclk"] + ] + sw_pin_params = [ + ppins.lookup_pin(config.get(name), share_type=name) for name in sw_pin_names + ] for pin_params in sw_pin_params: - if pin_params['chip'] != mcu: - raise ppins.error("%s: spi pins must be on same mcu" % ( - config.get_name(),)) - sw_pins = tuple([pin_params['pin'] for pin_params in sw_pin_params]) + if pin_params["chip"] != mcu: + raise ppins.error( + "%s: spi pins must be on same mcu" % (config.get_name(),) + ) + sw_pins = tuple([pin_params["pin"] for pin_params in sw_pin_params]) bus = None else: - bus = config.get('spi_bus', None) + bus = config.get("spi_bus", None) sw_pins = None # Create MCU_SPI object return MCU_SPI(mcu, bus, pin, mode, speed, sw_pins, cs_active_high) @@ -153,6 +186,7 @@ def MCU_SPI_from_config(config, mode, pin_option="cs_pin", # I2C ###################################################################### + # Helper code for working with devices connected to an MCU via an I2C bus class MCU_I2C: def __init__(self, mcu, bus, addr, speed, sw_pins=None): @@ -168,78 +202,97 @@ class MCU_I2C: self.config_fmt = ( "i2c_set_software_bus oid=%d" " scl_pin=%s sda_pin=%s rate=%d address=%d" - % (self.oid, sw_pins[0], sw_pins[1], speed, addr)) + % (self.oid, sw_pins[0], sw_pins[1], speed, addr) + ) self.config_fmt_ticks = ( "i2c_set_sw_bus oid=%d" " scl_pin=%s sda_pin=%s pulse_ticks=%%d address=%d" - % (self.oid, sw_pins[0], sw_pins[1], addr)) + % (self.oid, sw_pins[0], sw_pins[1], addr) + ) else: - self.config_fmt = ( - "i2c_set_bus oid=%d i2c_bus=%%s rate=%d address=%d" - % (self.oid, speed, addr)) + self.config_fmt = "i2c_set_bus oid=%d i2c_bus=%%s rate=%d address=%d" % ( + self.oid, + speed, + addr, + ) self.cmd_queue = self.mcu.alloc_command_queue() self.mcu.register_config_callback(self.build_config) self.i2c_write_cmd = self.i2c_read_cmd = None + def get_oid(self): return self.oid + def get_mcu(self): return self.mcu + def get_i2c_address(self): return self.i2c_address + def get_command_queue(self): return self.cmd_queue + def build_config(self): - if '%' in self.config_fmt: + if "%" in self.config_fmt: bus = resolve_bus_name(self.mcu, "i2c_bus", self.bus) self.config_fmt = self.config_fmt % (bus,) if self.config_fmt_ticks: - if self.mcu.try_lookup_command("i2c_set_sw_bus oid=%c" - " scl_pin=%u sda_pin=%u" - " pulse_ticks=%u address=%u"): - pulse_ticks = self.mcu.seconds_to_clock(1./self.speed/2) + if self.mcu.try_lookup_command( + "i2c_set_sw_bus oid=%c" + " scl_pin=%u sda_pin=%u" + " pulse_ticks=%u address=%u" + ): + pulse_ticks = self.mcu.seconds_to_clock(1.0 / self.speed / 2) self.config_fmt = self.config_fmt_ticks % (pulse_ticks,) self.mcu.add_config_cmd(self.config_fmt) self.i2c_write_cmd = self.mcu.lookup_command( - "i2c_write oid=%c data=%*s", cq=self.cmd_queue) + "i2c_write oid=%c data=%*s", cq=self.cmd_queue + ) self.i2c_read_cmd = self.mcu.lookup_query_command( "i2c_read oid=%c reg=%*s read_len=%u", - "i2c_read_response oid=%c response=%*s", oid=self.oid, - cq=self.cmd_queue) + "i2c_read_response oid=%c response=%*s", + oid=self.oid, + cq=self.cmd_queue, + ) + def i2c_write(self, data, minclock=0, reqclock=0): if self.i2c_write_cmd is None: # Send setup message via mcu initialization data_msg = "".join(["%02x" % (x,) for x in data]) - self.mcu.add_config_cmd("i2c_write oid=%d data=%s" % ( - self.oid, data_msg), is_init=True) + self.mcu.add_config_cmd( + "i2c_write oid=%d data=%s" % (self.oid, data_msg), is_init=True + ) return - self.i2c_write_cmd.send([self.oid, data], - minclock=minclock, reqclock=reqclock) + self.i2c_write_cmd.send([self.oid, data], minclock=minclock, reqclock=reqclock) + def i2c_write_wait_ack(self, data, minclock=0, reqclock=0): - self.i2c_write_cmd.send_wait_ack([self.oid, data], - minclock=minclock, reqclock=reqclock) + self.i2c_write_cmd.send_wait_ack( + [self.oid, data], minclock=minclock, reqclock=reqclock + ) + def i2c_read(self, write, read_len, retry=True): return self.i2c_read_cmd.send([self.oid, write, read_len], retry) + def MCU_I2C_from_config(config, default_addr=None, default_speed=100000): # Load bus parameters printer = config.get_printer() - i2c_mcu = mcu.get_printer_mcu(printer, config.get('i2c_mcu', 'mcu')) - speed = config.getint('i2c_speed', default_speed, minval=100000) + i2c_mcu = mcu.get_printer_mcu(printer, config.get("i2c_mcu", "mcu")) + speed = config.getint("i2c_speed", default_speed, minval=100000) if default_addr is None: - addr = config.getint('i2c_address', minval=0, maxval=127) + addr = config.getint("i2c_address", minval=0, maxval=127) else: - addr = config.getint('i2c_address', default_addr, minval=0, maxval=127) + addr = config.getint("i2c_address", default_addr, minval=0, maxval=127) # Determine pin from config ppins = config.get_printer().lookup_object("pins") - if config.get('i2c_software_scl_pin', None) is not None: - sw_pin_names = ['i2c_software_%s_pin' % (name,) - for name in ['scl', 'sda']] - sw_pin_params = [ppins.lookup_pin(config.get(name), share_type=name) - for name in sw_pin_names] - sw_pins = tuple([pin_params['pin'] for pin_params in sw_pin_params]) + if config.get("i2c_software_scl_pin", None) is not None: + sw_pin_names = ["i2c_software_%s_pin" % (name,) for name in ["scl", "sda"]] + sw_pin_params = [ + ppins.lookup_pin(config.get(name), share_type=name) for name in sw_pin_names + ] + sw_pins = tuple([pin_params["pin"] for pin_params in sw_pin_params]) bus = None else: - bus = config.get('i2c_bus', None) + bus = config.get("i2c_bus", None) sw_pins = None # Create MCU_I2C object return MCU_I2C(i2c_mcu, bus, addr, speed, sw_pins) @@ -249,38 +302,48 @@ def MCU_I2C_from_config(config, default_addr=None, default_speed=100000): # Bus synchronized digital outputs ###################################################################### + # Helper code for a gpio that updates on a cmd_queue class MCU_bus_digital_out: def __init__(self, mcu, pin_desc, cmd_queue=None, value=0): self.mcu = mcu self.oid = mcu.create_oid() - ppins = mcu.get_printer().lookup_object('pins') + ppins = mcu.get_printer().lookup_object("pins") pin_params = ppins.lookup_pin(pin_desc) - if pin_params['chip'] is not mcu: - raise ppins.error("Pin %s must be on mcu %s" % ( - pin_desc, mcu.get_name())) - mcu.add_config_cmd("config_digital_out oid=%d pin=%s value=%d" - " default_value=%d max_duration=%d" - % (self.oid, pin_params['pin'], value, value, 0)) + if pin_params["chip"] is not mcu: + raise ppins.error("Pin %s must be on mcu %s" % (pin_desc, mcu.get_name())) + mcu.add_config_cmd( + "config_digital_out oid=%d pin=%s value=%d" + " default_value=%d max_duration=%d" + % (self.oid, pin_params["pin"], value, value, 0) + ) mcu.register_config_callback(self.build_config) if cmd_queue is None: cmd_queue = mcu.alloc_command_queue() self.cmd_queue = cmd_queue self.update_pin_cmd = None + def get_oid(self): return self.oid + def get_mcu(self): return self.mcu + def get_command_queue(self): return self.cmd_queue + def build_config(self): self.update_pin_cmd = self.mcu.lookup_command( - "update_digital_out oid=%c value=%c", cq=self.cmd_queue) + "update_digital_out oid=%c value=%c", cq=self.cmd_queue + ) + def update_digital_out(self, value, minclock=0, reqclock=0): if self.update_pin_cmd is None: # Send setup message via mcu initialization - self.mcu.add_config_cmd("update_digital_out oid=%c value=%c" - % (self.oid, not not value)) + self.mcu.add_config_cmd( + "update_digital_out oid=%c value=%c" % (self.oid, not not value) + ) return - self.update_pin_cmd.send([self.oid, not not value], - minclock=minclock, reqclock=reqclock) + self.update_pin_cmd.send( + [self.oid, not not value], minclock=minclock, reqclock=reqclock + ) |