pins.py

# Pin name handling
#
# Copyright (C) 2016-2021  Kevin O'Connor <kevin@koconnor.net>
#
# This file may be distributed under the terms of the GNU GPLv3 license.
import re


class error(Exception):
    pass


######################################################################
# Command translation
######################################################################

re_pin = re.compile(r"(?P<prefix>[ _]pin=)(?P<name>[^ ]*)")


class PinResolver:
    def __init__(self, validate_aliases=True):
        self.validate_aliases = validate_aliases
        self.reserved = {}
        self.aliases = {}
        self.active_pins = {}

    def reserve_pin(self, pin, reserve_name):
        if pin in self.reserved and self.reserved[pin] != reserve_name:
            raise error(
                "Pin %s reserved for %s - can't reserve for %s"
                % (pin, self.reserved[pin], reserve_name)
            )
        self.reserved[pin] = reserve_name

    def alias_pin(self, alias, pin):
        if alias in self.aliases and self.aliases[alias] != pin:
            raise error(
                "Alias %s mapped to %s - can't alias to %s"
                % (alias, self.aliases[alias], pin)
            )
        if [c for c in "^~!:" if c in pin] or "".join(pin.split()) != pin:
            raise error("Invalid pin alias '%s'\n" % (pin,))
        if pin in self.aliases:
            pin = self.aliases[pin]
        self.aliases[alias] = pin
        for existing_alias, existing_pin in self.aliases.items():
            if existing_pin == alias:
                self.aliases[existing_alias] = pin

    def update_command(self, cmd):
        def pin_fixup(m):
            name = m.group("name")
            pin_id = self.aliases.get(name, name)
            if (
                name != self.active_pins.setdefault(pin_id, name)
                and self.validate_aliases
            ):
                raise error(
                    "pin %s is an alias for %s"
                    % (name, self.active_pins[pin_id])
                )
            if pin_id in self.reserved:
                raise error(
                    "pin %s is reserved for %s" % (name, self.reserved[pin_id])
                )
            return m.group("prefix") + str(pin_id)

        return re_pin.sub(pin_fixup, cmd)


######################################################################
# Pin to chip mapping
######################################################################


class PrinterPins:
    error = error

    def __init__(self):
        self.chips = {}
        self.active_pins = {}
        self.pin_resolvers = {}
        self.allow_multi_use_pins = {}

    def parse_pin(self, pin_desc, can_invert=False, can_pullup=False):
        desc = pin_desc.strip()
        pullup = invert = 0
        if can_pullup and (desc.startswith("^") or desc.startswith("~")):
            pullup = 1
            if desc.startswith("~"):
                pullup = -1
            desc = desc[1:].strip()
        if can_invert and desc.startswith("!"):
            invert = 1
            desc = desc[1:].strip()
        if ":" not in desc:
            chip_name, pin = "mcu", desc
        else:
            chip_name, pin = [s.strip() for s in desc.split(":", 1)]
        if chip_name not in self.chips:
            raise error("Unknown pin chip name '%s'" % (chip_name,))
        if [c for c in "^~!:" if c in pin] or "".join(pin.split()) != pin:
            format = ""
            if can_pullup:
                format += "[^~] "
            if can_invert:
                format += "[!] "
            raise error(
                "Invalid pin description '%s'\n"
                "Format is: %s[chip_name:] pin_name" % (pin_desc, format)
            )
        pin_params = {
            "chip": self.chips[chip_name],
            "chip_name": chip_name,
            "pin": pin,
            "invert": invert,
            "pullup": pullup,
        }
        return pin_params

    def lookup_pin(
        self, pin_desc, can_invert=False, can_pullup=False, share_type=None
    ):
        pin_params = self.parse_pin(pin_desc, can_invert, can_pullup)
        pin = pin_params["pin"]
        share_name = "%s:%s" % (pin_params["chip_name"], pin)
        if share_name in self.active_pins:
            share_params = self.active_pins[share_name]
            if share_name in self.allow_multi_use_pins:
                pass
            elif share_type is None or share_type != share_params["share_type"]:
                raise error("pin %s used multiple times in config" % (pin,))
            elif (
                pin_params["invert"] != share_params["invert"]
                or pin_params["pullup"] != share_params["pullup"]
            ):
                raise error("Shared pin %s must have same polarity" % (pin,))
            return share_params
        pin_params["share_type"] = share_type
        self.active_pins[share_name] = pin_params
        return pin_params

    def setup_pin(self, pin_type, pin_desc):
        can_invert = pin_type in ["endstop", "digital_out", "pwm"]
        can_pullup = pin_type in ["endstop"]
        pin_params = self.lookup_pin(pin_desc, can_invert, can_pullup)
        return pin_params["chip"].setup_pin(pin_type, pin_params)

    def reset_pin_sharing(self, pin_params):
        share_name = "%s:%s" % (pin_params["chip_name"], pin_params["pin"])
        del self.active_pins[share_name]

    def get_pin_resolver(self, chip_name):
        if chip_name not in self.pin_resolvers:
            raise error("Unknown chip name '%s'" % (chip_name,))
        return self.pin_resolvers[chip_name]

    def register_chip(self, chip_name, chip):
        chip_name = chip_name.strip()
        if chip_name in self.chips:
            raise error("Duplicate chip name '%s'" % (chip_name,))
        self.chips[chip_name] = chip
        self.pin_resolvers[chip_name] = PinResolver()

    def allow_multi_use_pin(self, pin_desc):
        pin_params = self.parse_pin(pin_desc)
        share_name = "%s:%s" % (pin_params["chip_name"], pin_params["pin"])
        self.allow_multi_use_pins[share_name] = True


def add_printer_objects(config):
    config.get_printer().add_object("pins", PrinterPins())