klipper-macros/extras/bdwidth_status.py

# Expose bdwidth's motion counter and width reading to gcode macros.
#
# bdwidth.py replaces its own get_status() with the runout helper's, so
# macros only see filament_detected/enabled. This read-only wrapper looks
# up the existing [bdwidth <name>] object and reports the values it
# already keeps. It never talks to the sensor, so bdwidth.py stays
# untouched and updates to bdwidth do not overwrite this file.
#
# Install: copy to ~/klipper/klippy/extras/bdwidth_status.py
#
# Config:
#   [bdwidth_status]
#   sensor: fila_width_0   # name after "bdwidth" in [bdwidth fila_width_0]
#
# Status (printer.bdwidth_status):
#   total_move     - motion counter since Klipper start (raw counts, signed)
#   last_motion    - counts seen in the last sample
#   diameter       - last width reading in mm (0.0 on a bad reading)
#   name           - the sensor name, for SET_BDWIDTH NAME=...
#   reading        - True while bdwidth is polling the sensor (after
#                    SET_BDWIDTH ... ENABLE..., until DISABLE)
#   enabled        - runout response on (SET_FILAMENT_SENSOR ENABLE=1/0);
#                    says nothing about whether bdwidth is reading
#   active         - "motion", "width", "all" or "disable" (SET_BDWIDTH). After
#                    a Klipper restart it shows the config value, but bdwidth
#                    only starts reading on the first SET_BDWIDTH ... ENABLE
#   linear_motion  - counts per mm (motion_linear_coefficient)
#   sample_time    - seconds between sensor reads
#   present        - False when no [bdwidth <sensor>] is configured; the
#                    other fields then hold neutral defaults, so macros keep
#                    working with bdwidth removed from the config
#   usb_ok         - False while the USB connection is down (port: usb)
#   usb_errors     - USB read errors since Klipper start
#
# USB guard (port: usb): bdwidth reads the sensor from a reactor timer without
# any error handling, so one USB dropout of its CH340 (write failed: [Errno 5]
# Input/output error) shut down Klipper mid print. This module wraps the
# sensor's Read_bdwidth at connect: on a USB error it closes the port, reports
# it and returns False, after which bdwidth tries again 10 s later; the next
# reads reopen the port (the /dev/serial/by-id path returns when the device
# reconnects). bdwidth.py itself is not changed.


import logging


class BDWidthStatus:
    def __init__(self, config):
        self.printer = config.get_printer()
        self.name = config.get('sensor', 'fila_width_0')
        self.sensor_name = 'bdwidth ' + self.name
        self.sensor = None
        self.usb_ok = True
        self.usb_errors = 0
        self._read = None
        self.printer.register_event_handler('klippy:connect',
                                            self._handle_connect)

    def _handle_connect(self):
        self.sensor = self.printer.lookup_object(self.sensor_name, None)
        if self.sensor is None:
            logging.info("bdwidth_status: no [%s] configured", self.sensor_name)
            return
        if (getattr(self.sensor, 'port', None) == 'usb'
                and callable(getattr(self.sensor, 'Read_bdwidth', None))
                and getattr(self.sensor, 'usb', None) is not None):
            self._read = self.sensor.Read_bdwidth
            self.sensor.Read_bdwidth = self._guarded_read
            logging.info("bdwidth_status: USB guard active for %s", self.name)

    def _respond(self, msg):
        logging.warning(msg)
        try:
            self.printer.lookup_object('gcode').respond_info(msg)
        except Exception:
            pass

    def _guarded_read(self):
        # Called from bdwidth's reactor timer: must never raise
        usb = self.sensor.usb
        try:
            if not usb.is_open:
                usb.open()
            result = self._read()
        except Exception as e:
            self.usb_errors += 1
            try:
                usb.close()
            except Exception:
                pass
            if self.usb_ok:
                self.usb_ok = False
                self._respond("bdwidth %s: USB error (%s), print continues "
                              "without width readings, retrying"
                              % (self.name, e))
            return False
        if not self.usb_ok:
            self.usb_ok = True
            self._respond("bdwidth %s: USB connection restored after %d "
                          "errors" % (self.name, self.usb_errors))
        return result

    def get_status(self, eventtime):
        s = self.sensor
        if s is None:
            return {'name': self.name, 'present': False, 'reading': False,
                    'enabled': False, 'total_move': 0, 'last_motion': 0,
                    'diameter': 0., 'active': 'disable', 'linear_motion': 42.8,
                    'sample_time': 1., 'usb_ok': True, 'usb_errors': 0}
        # bdwidth's own get_status() is the runout helper's, which carries
        # the SET_FILAMENT_SENSOR enable flag
        try:
            enabled = bool(s.get_status(eventtime).get('enabled', True))
        except Exception:
            enabled = True
        # bdwidth reads from a reactor timer that only SET_BDWIDTH ... ENABLE
        # starts and DISABLE stops; is_active alone does not show it
        timer = getattr(s, 'extrude_factor_update_timer', None)
        never = self.printer.get_reactor().NEVER
        reading = (timer is not None and getattr(timer, 'waketime', never) != never
                   and 'disable' not in str(getattr(s, 'is_active', 'disable')))
        return {
            'name': self.name,
            'present': True,
            'reading': reading,
            'enabled': enabled,
            'total_move': getattr(s, 'actual_total_move', 0),
            'last_motion': getattr(s, 'lastMotionReading', 0),
            'diameter': round(getattr(s, 'lastFilamentWidthReading', 0.), 3),
            'active': str(getattr(s, 'is_active', 'disable')),
            'linear_motion': getattr(s, 'linear_motion', 42.8),
            'sample_time': getattr(s, 'sample_time', 1.),
            'usb_ok': self.usb_ok,
            'usb_errors': self.usb_errors,
        }


def load_config(config):
    return BDWidthStatus(config)