DigiLib/pin/__init__.py

#!/usr/bin/env python3.5
# Copyright 2017 Digital
#
# This file is part of DigiLib.
#
# DigiLib is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# DigiLib is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with DigiLib.  If not, see <http://www.gnu.org/licenses/>.

# Python modules
import logging
import threading
import traceback
# Third party modules
import blinker
import curio
import digilib.network

log = logging.getLogger(__name__+"")
lgpio = logging.getLogger(__name__+".gpio")

_pins_for_cleanup = set()

# Error messages
ERROR_TAKES_ARGUMENTS = "{} takes {} {} argument(s): {}"
# respond(ERROR_TAKES_ARGUMENTS.format(
#         command, "one", "positional", "<name>"))

# def _parse_to_int_list(parse):
#     if type(parse) is list:
#         return parse
#     elif type(parse) is int:
#         return [parse]
#     elif type(parse) is bool:
#         return [int(parse)]
#     elif type(parse) is not str:
#         raise ValueError("Failed to parse value {}".format(parse))
#     if parse.lower() in ["true","high"]:
#         return [1]
#     elif parse.lower() in ["false","low"]:
#         return [0]
#     parse = parse.split(".")
#     for part in parse.copy():
#         parse.remove(part)
#         parse.append(int(part))
#     return parse
#
# def _setup()
#
# def _output(pins,state):
#     lgpio.debug("setting pin(s) {:0>2} to value {}".format(
#         pins,state))
#     if type(pins) is int:
#         pins = [pins]
#     _pins_for_cleanup.update(pins)
#     gpio.output(pins,state)
#
# def _read(self,pin):
#     state = gpio.read(pin)
#     lgpio.debug("reading pin {}: {}".format(
#         pin,state))

# class FakeGPIO(object):
#     OUT = "out"
#     IN = "in"
#     BCM = "broadcom"
#     pin_values = {}
#     def cleanup(self,*args):
#         lgpio.debug("cleanup with these args '{}'!".format(args))
#     def setup(self,pins,value):
#         lgpio.debug("pin(s) {} set to {}".format(pins,value))
#     def setmode(self,mode):
#         lgpio.debug("mode set to {}".format(mode))
#     def output(self,pins,value):
#         lgpio.debug("setting pin(s) {} to value {}".format(
#             pins, value
#         ))
#         if type(pins) == int:
#             pins = [pins]
#         for p in pins:
#             self.pin_values[p] = value
#     def read(self,pin):
#         return self.pin_values[pin]
#
# try:
#     import RPi.GPIO as gpio
# except:
#     # print("Using FakeGPIO because RPi.GPIO was not found")
#     gpio = FakeGPIO()
#
# def call_gpio_cleanup(*args):
#     log.info(args)
#     gpio.cleanup(_pins_for_cleanup)


class GPIOWrapper(object):
    gpio = None
    OUT = "out"
    IN = "in"
    BCM = "broadcom"
    pin_values = {}

    def __init__(self):
        super(GPIOWrapper,self).__init__()
        self.load_rpi_gpio()
        lgpio.debug("setting pin numbering to broadcom")
        if self.gpio:
            self.gpio.setmode(self.gpio.BCM)
        sig_exit = blinker.signal("global-exit")
        sig_exit.connect(self.cleanup)

    def cleanup(self,*args):
        lgpio.debug("cleanup! ({})".format(args))
        if self.gpio:
            self.gpio.cleanup(_pins_for_cleanup)

    def load_rpi_gpio(self):
        try:
            self.gpio = __import__("RPi.GPIO")
        except ImportError as e:
            lgpio.debug("failed to import RPi.GPIO")
        except Exception as e:
            lgpio.debug("unknown error occured", exc_info=e)
        finally:
            self.gpio = False
            self.OUT = getattr(self.gpio,self.OUT)
            self.IN = getattr(self.gpio,self.IN)
            self.BCM = getattr(self.gpio,self.BCM)

    def output(pins,state,*args):
        lgpio.debug("setting pin(s) {:0>2} to value {}".format(
            pins,state))
        if type(pins) is int:
            pins = [pins]
        _pins_for_cleanup.update(pins)
        if self.gpio:
            self.gpio.output(pins,state)

    def parse_to_int_list(parse,*args):
        if type(parse) is list:
            return parse
        elif type(parse) is int:
            return [parse]
        elif type(parse) is bool:
            return [int(parse)]
        elif type(parse) is not str:
            raise ValueError("Failed to parse value {}".format(parse))
        if parse.lower() in ["true","high"]:
            return [1]
        elif parse.lower() in ["false","low"]:
            return [0]
        parse = parse.split(".")
        for part in parse.copy():
            parse.remove(part)
            parse.append(int(part))
        return parse

    def read(self,pin,*args):
        if self.gpio:
            state = self.gpio.read(pin)
        else:
            state = -1
        lgpio.debug("reading pin {:0>2}: {}".format(
            pin,state))
        return state

    def setup(self,pins,value,*args):
        lgpio.debug("setting pin(s) {} to {}".format(pins,value))
        if self.gpio:
            self.gpio.setup(pins,value)

class PinBase(object):
    """PinBase is the base class for all classes representing a gpio pin"""
    pin_number = None
    value = None
    def __init__(self,pin_number,mode):
        super(PinBase,self).__init__()
        self.pin_number = pin_number
        self.value = self.value_low

    def output(self,value):
        [value] = _parse_to_int_list(value)
        self.value = value
        _output(self.pin_number,value)

    def read(self):
        value = _read(self.pin_number,value)
        return value

class DigitalPin(PinBase):
    value_high = True
    value_low = False
    def __init__(self,pin_number,mode):
        super(DigitalPin,self).__init__(pin_number,mode)

class AnalogPin(PinBase):
    value_high = 1
    value_low = 0
    def __init__(self,pin_number):
        super(AnalogPin,self).__init__(pin_number)

class PinControllerBase(object):
    """docstring for PinControllerBase.
    PinControllerBase is the base class for all classes controlling one or more physical devices connected to a gpio header

    """
    pins = []
    def __init__(self,pins):
        super(PinControllerBase, self).__init__()
        self.pins.extend(pins)
    # def make_digital_pin(self,*args):
    #     return DigitalPin(*args)
    # def make_analog_pin(self,*args):
    #     return AnalogPin(*args)

class PinAPIBase(object):
    """docstring for PinAPI.
    PinAPIBase is the base class for all classes providing an api to multiple
    PinController.
    """
    controllers = []
    def __init__(self):
        super(PinAPIBase, self).__init__()

class PCEngine(PinControllerBase):
    """Test Class"""
    max_speed=1
    speed = 0
    turn_on_speed = 1
    is_on = False
    def __init__(self,pin_on_off,pin_analog):
        super(PCEngine, self).__init__([pin_on_off,pin_analog])
        # self.pins.append(pin_on_off)
        # self.pins.append(pin_analog)
        self.pin_on_off = DigitalPin(pin_on_off,gpio.OUT)
        self.pin_analog = DigitalPin(pin_analog,gpio.OUT)
        # self.pins.append(self.pin_on_off)
        # self.pins.append(self.pin_analog)
        # gpio.setup(self.pins,gpio.OUT)
        # gpio.setup(self.pin_analog,gpio.OUT)
        self.set_speed(1)
        self.set_state(1)
    def set_speed(self,speed):
        self.pin_analog.output(speed)
        self.speed = speed
    def set_state(self,state):
        """state is integer and specifies if the engine should be turned on (1) or turned off (0) """
        if state == self.is_on:
            return
        # self.set_speed(speed)
        self.pin_on_off.output(state)
        self.is_on = state

class EnginesController(PinAPIBase):
    engine_left = None
    engine_right = None
    action_in_process = False
    def __init__(self,left,right):
        super(EnginesController,self).__init__()
        self.engine_right = self.make_engine(*right)
        self.engine_left = self.make_engine(*left)
    def __enter__(self):
        if self.action_in_process:
            lgpio.debug("action already in process, not entering")
            return False
    def make_engine(self,*args):
        return digilib.pin.PCEngine(*args)
    def set_engine_state(self,args=[],command=None,respond=None):
        if len(args) != 2:
            respond("missing argument(s): <engine> <state>")
            return
        [engine,state] = args
        if engine == "right":
            self.engine_right.set_state(state)
        elif engine == "left":
            self.engine_left.set_state(state)
        else:
            respond("unknown engine. use 'left' or 'right'")
    def set_engine_speed(self,args=[],command=None,respond=None):
        if len(args) != 2:
            respond("missing argument(s): <engine> <speed>")
            return
        [engine,speed] = args
        if engine == "right":
            self.engine_right.set_speed(speed)
        elif engine == "left":
            self.engine_left.set_speed(speed)
        else:
            respond("unknown engine. use 'left' or 'right'")
    async def turn(self,args=[],command=None,respond=None):
        if len(args) != 1:
            respond("one missing argument: direction")
            return
        [direction] = args
        # lgpio.debug(threading.current_thread())
        # with self ad
        lgpio.info("turning {}".format(direction))
        right_state = self.engine_right.is_on
        right_speed = self.engine_right.speed
        left_state = self.engine_left.is_on
        left_speed = self.engine_left.speed
        if direction == "right":
            self.engine_right.set_state(False)
            self.engine_left.set_state(True)
            self.engine_left.set_speed(1)
        elif direction == "left":
            self.engine_right.set_state(True)
            self.engine_right.set_speed(1)
            self.engine_left.set_state(False)
        await curio.sleep(2)
        # time.sleep(2)
        self.engine_right.set_state(right_state)
        self.engine_right.set_speed(right_speed)
        self.engine_left.set_state(left_state)
        self.engine_left.set_speed(left_speed)
        lgpio.info("done turning {}".format(direction))
        # set engines to previous values


class LED(DigitalPin):
    def __init__(self,pin):
        super(DigitalPin,self).__init__(pin,gpio.OUT)
        self.on()
    def on(self,args=[],command=None,respond=None):
        self.output(True)
    def off(self,args=[],command=None,respond=None):
        self.output(False)
    def set(self,args=[],command=None,respond=None):
        if len(args) != 1:
            respond("one missing argument: state")
            return
        [state] = args
        self.output(state)

class LEDRGB(PinControllerBase):
    def __init__(self,pin_red,pin_green):
        super(LEDRGB,self).__init__([pin_red,pin_green])
        self.pin_red = DigitalPin(pin_red,gpio.OUT)
        self.pin_green = DigitalPin(pin_green,gpio.OUT)
        self.red()
    def red(self,args=[],command=None,respond=None):
        if len(args) > 1:
            respond("{} takes one optinal agument: <state>".foramt(command))
            return
        elif len(args) == 1:
            state = int(*args)
        else:
            state = 1
        self.pin_red.output(state)
    def green(self,args=[],command=None,respond=None):
        if len(args) > 1:
            respond("{} takes one optinal agument: <state>".foramt(command))
            return
        elif len(args) == 1:
            state = int(*args)
        else:
            state = 1
        self.pin_green.output(state)

class DebugPinController(PinControllerBase):
    def output(self,args=[],command=None,respond=None):
        if len(args) != 2:
            respond(ERROR_TAKES_ARGUMENTS.format(
                    command, "two", "positional", "<name>"))
            return False
        pins = _parse_to_int_list(args[0])
        [state] = _parse_to_int_list(args[1])
        _output(pins,state)
    def output(self,args=[],command=None,respond=None):
        if len(args) != 2:
            respond(ERROR_TAKES_ARGUMENTS.format(
                    command, "two", "positional", "<name>"))
            return False
        pins = _parse_to_int_list(args[0])
        [state] = _parse_to_int_list(args[1])
        return _read(pins,state)

if __name__ == "__main__":
    pass






























#