Блог

У меня есть синглтон, в котором я хочу получать данные от службы. Поскольку я новичок в python, единственный способ, которым я мог это понять, заключался в том, что служба отправляет данные в синглтон через указатель метода. Я использую поток, чтобы помешать службе заблокировать синглтон с его бесконечным циклом.

Фактические обстоятельства — это то, что я хочу продлить Издевательство.GPIO(макет реализации RPi.GPIO), https://github.com/codenio/Mock .GPIO, чтобы иметь возможность имитировать нажатие кнопки на реальном RPi. Синглтон представляет собой упрощенную плату RPi.

Я хочу вызвать службу сокетов со следующим echo -n "22:HI" > /dev/tcp/127.0.0.1/5566

Я хочу, чтобы это привело к

 Service __init__
piBoardCallback
22:HI
22
HI
<something that represents event>
 

Код фактически выполняется прямо сейчас.

Вместо этого я получаю 5 ошибок, что и является проблемой на данный момент!!!

     Exception in thread Thread-1:
    Traceback (most recent call last):
      File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
        self.run()
      File "/usr/lib/python2.7/threading.py", line 754, in run
        self.__target(*self.__args, **self.__kwargs)
      File "/Extra/home/max/DEV/Code/workspaces/2019/Mock.GPIO/Mock/PiBoard.py", line 130, in run
        self.svc.listen(self.piBoardCallback)
      File "/Extra/home/max/DEV/Code/workspaces/2019/Mock.GPIO/Mock/PiBoard.py", line 108, in listen
        self.serviceThreadCallback(data)
    TypeError: piBoardCallback() takes exactly 2 arguments (1 given)
 

Я теряюсь, когда дело доходит до self, class, class(тип) и class (объект).

Пожалуйста, укажите мне правильное направление.

Ниже приведены занятия, которые у меня есть на данный момент. Они ни в коей мере не закончены.

Если у вас есть какое-то другое решение для передачи изменения сигнала в метод обратного вызова, зарегистрированный с помощью метода add_event_detect, я открыт для предложений.

Ссылка на оригинальный Макет.Код GPIO, https://github.com/codenio/Mock.GPIO/blob/master/Mock/GPIO.py

Мой источник

     import GPIO as GPIO
    import threading
    import time
    import socket as sk
    
    class Board():
        channelConfigs = {}
        channelEvents = {}
        __instance = None
        global serviceThread
        global piBoardCallback
        
        @staticmethod 
        def getInstance():
            """ Static access method. """
            if Board.__instance == None:
                Board()
            return Board.__instance
        
        def __init__(self):
            if Board.__instance != None:
                raise Exception("This class is a singleton!")
            else:
                self.serviceThread = ServiceThread()
                self.serviceThread.piBoardCallback = piBoardCallback
                self.serviceThread.threadify()
                
                Board.__instance = self
                
               
        def piBoardCallback(self,val):
            print "piBoardCallback"
            # This assumes that val is in format {channel:[HI|LOW]}
            x = val.split(":")
            print(val)
            channel = x[0]
            edge = x[1]
            print(channel)
            print(edge)
            event = self.channelEvents[channel]
            print(event)
        
        def setChannelConfig(self, channel):
            if channel != None:
                self.channelConfigs[channel.chanel] = channel
            
        def setChannelEvent(self, channel, edge, eventCallback):
            if channel != None:
                event = Event(edge,eventCallback)
                self.channelEvents[channel] = Event(edge,eventCallback)
       
    # class Event(object):
    class Event:
        '''
        classdocs
        '''
    
        global eventCallback
        global edge
    
        def __init__(self,edge,eventCallback):
            '''
            Constructor
            '''
            self.edge = edge
            self.eventCallback = eventCallback 
        
        def getEventCallback():
            return self.eventCallback
       
        def setEventCallback(eventCallback):
            self.eventCallback = eventCallback
       
    class Service:
    
        global serviceThreadCallback
     
        def __init__(self):
            print "Service __init__"
            
        def listen(self,serviceThreadCallback):
            self.serviceThreadCallback = serviceThreadCallback
            connection = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
            connection.setsockopt(sk.SOL_SOCKET, sk.SO_REUSEADDR, 1)
            connection.bind(('0.0.0.0', 5566))
            connection.listen(10)
            while True:
                current_connection, address = connection.accept()
                while True:
                    data = current_connection.recv(2048)
        
                    if data == 'quit\n':
                        current_connection.shutdown(1)
                        current_connection.close()
                        break
        
                    elif data == 'stop\n':
                        current_connection.shutdown(1)
                        current_connection.close()
                        exit()
        
                    elif data:
                        self.serviceThreadCallback(data)
                        
                    else:
                        break         
        
        def setCallback(self, serviceThreadCallback):
            self.serviceThreadCallback = serviceThreadCallback
            
        def getCallback(serviceThreadCallback):
           return self.serviceThreadCallback
            
    class ServiceThread:
    
        global piBoardCallback
        global thread
        global svc
    
        def __init__(self, interval=1):
            self.interval = interval
    
        def run(self):
                self.svc = Service()
                self.svc.listen(self.piBoardCallback)
    
        def setPiBoardCallback(piBoardCallback):
            self.piBoardCallback = piBoardCallback
     
        def getPiBoardCallback():
            return self.piBoardCallback
            
        def threadify(self):
            self.thread = threading.Thread(target=self.run)
            self.thread.daemon = True  # Daemonize thread
            self.thread.start()  # Start the execution
            
        def serviceThreadCallback(val):
            self.piBoardCallback(val)
    
    def ext_callback(val):
        print "ext_callback"
        print val        
                    
    if __name__ == '__main__': 
         
        try:
            while True:
                GPIO.setmode(GPIO.BCM)
                GPIO.setup(22, GPIO.IN, 0, GPIO.PUD_UP)
                GPIO.add_event_detect(22, GPIO.FALLING, ext_callback, bouncetime=1500)
                time.sleep(1000)
        except KeyboardInterrupt:
            pass      
                  
   
 

Мой код GPIO

     """
    Mock Library for RPi.GPIO
    """
    
    import time
    import logging
    import os
    # import yaml
    import PiBoard
    
    logger = logging.getLogger(__name__)
    
    log_level = os.getenv('LOG_LEVEL')
    
    if log_level is not None:
        if log_level == "Info":
            logger.setLevel(logging.INFO)
        if log_level == "Debug":
            logger.setLevel(logging.DEBUG)
        if log_level == "Warning":
            logger.setLevel(logging.WARNING)
        if log_level == "Error":
            logger.setLevel(logging.ERROR)
        if log_level == "Critical":
            logger.setLevel(logging.CRITICAL)
    else:
        logger.setLevel(logging.ERROR)
    
    stream_formatter = logging.Formatter('%(asctime)s:%(levelname)s: %(message)s')
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(stream_formatter)
    logger.addHandler(stream_handler)
    
    BCM = 11
    BOARD = 10
    BOTH = 33
    FALLING = 32
    HARD_PWM = 43
    HIGH = 1
    I2C = 42
    IN = 1
    LOW = 0
    OUT = 0
    PUD_DOWN = 21
    PUD_OFF = 20
    PUD_UP = 22
    RISING = 31
    RPI_INFO = {'MANUFACTURER': 'Sony', 'P1_REVISION': 3, 'PROCESSOR': 'BCM2837', 'RAM': '1G', 'REVISION': 'a020d3', 'TYPE': 'Pi 3 Model B '}
    RPI_REVISION = 3
    SERIAL = 40
    SPI = 41
    UNKNOWN = -1
    VERSION = '0.7.0'
    
    _mode = 0
    
    channel_config = {}
    
    #flags
    setModeDone = False
    
    class Channel:
        def __init__(self,channel, direction, initial=0,pull_up_down=PUD_OFF):
            self.chanel = channel
            self.direction = direction
            self.initial = initial
            self.current = initial
            self.pull_up_down = pull_up_down
    
    
    #GPIO LIBRARY Functions
    def setmode(mode):
        """
        Set up numbering mode to use for channels.
        BOARD - Use Raspberry Pi board numbers
        BCM   - Use Broadcom GPIO 00..nn numbers
        """
        
        board = getBoard()
        
        # GPIO = GPIO()
        time.sleep(1)
        if(mode == BCM):
            setModeDone = True
            _mode = mode
    
        elif (mode == BOARD):
            setModeDone = True
        else:
            setModeDone = False
    
    def getmode():
        """
        Get numbering mode used for channel numbers.
        Returns BOARD, BCM or None
        """
        board = getBoard()
        return _mode
    
    def setwarnings(flag):
        """
        Enable or disable warning messages
        """
        board = getBoard()
        logger.info("Set Warings as {}".format(flag))
    
    def setup(channel, direction, initial=0,pull_up_down=PUD_OFF):
        """
        Set up a GPIO channel or list of channels with a direction and (optional) pull/up down control
        channel        - either board pin number or BCM number depending on which mode is set.
        direction      - IN or OUT
        [pull_up_down] - PUD_OFF (default), PUD_UP or PUD_DOWN
        [initial]      - Initial value for an output channel
    
        """
        logger.info("setup channel : {} as {} with intial :{} and pull_up_dowm {}".format(channel,direction,initial,pull_up_down))
        board = getBoard()
        global channel_config
        channel_config[channel] = Channel(channel, direction, initial, pull_up_down)
        board.setChannelConfig(channel_config[channel])
    
    def output(channel, value):
        """
        Output to a GPIO channel or list of channels
        channel - either board pin number or BCM number depending on which mode is set.
        value   - 0/1 or False/True or LOW/HIGH
    
        """
        board = getBoard()
        logger.info("output channel : {} with value : {}".format(channel, value))
    
    def input(channel):
        """
        Input from a GPIO channel.  Returns HIGH=1=True or LOW=0=False
        channel - either board pin number or BCM number depending on which mode is set.
        """
        board = getBoard()
        logger.info("reading from chanel {}".format(channel))
    
    def wait_for_edge(channel,edge,bouncetime,timeout):
        """
        Wait for an edge.  Returns the channel number or None on timeout.
        channel      - either board pin number or BCM number depending on which mode is set.
        edge         - RISING, FALLING or BOTH
        [bouncetime] - time allowed between calls to allow for switchbounce
        [timeout]    - timeout in ms
        """
        board = getBoard()
        
        logger.info("waiting for edge : {} on channel : {} with bounce time : {} and Timeout :{}".format(edge,channel,bouncetime,timeout))
    
    
    def add_event_detect(channel,edge,callback,bouncetime):
        """
        Enable edge detection events for a particular GPIO channel.
        channel      - either board pin number or BCM number depending on which mode is set.
        edge         - RISING, FALLING or BOTH
        [callback]   - A callback function for the event (optional)
        [bouncetime] - Switch bounce timeout in ms for callback
        """
    #     board = 
        getBoard().setChannelEvent(channel, edge, callback)
        logger.info("Event detect added for edge : {} on channel : {} with bouce time : {} and callback {}".format(edge,channel,bouncetime,callback))
    
    def event_detected(channel):
        """
        Returns True if an edge has occurred on a given GPIO.  You need to enable edge detection using add_event_detect() first.
        channel - either board pin number or BCM number depending on which mode is set.
        """
        board = getBoard()
        logger.info("Waiting for even detection on channel :{}".format(channel))
    
    def add_event_callback(channel,callback):
        """
        Add a callback for an event already defined using add_event_detect()
        channel      - either board pin number or BCM number depending on which mode is set.
        callback     - a callback function
        """
        logger.info("Event Calback : {} added for channel : {}".format(callback,channel))
    
    def remove_event_detect(channel):
        """
        Remove edge detection for a particular GPIO channel
        channel - either board pin number or BCM number depending on which mode is set.
        """
        board = getBoard()
        logger.info("Event Detect Removed for channel : {}".format(channel))
    
    def gpio_function(channel):
        """
        Return the current GPIO function (IN, OUT, PWM, SERIAL, I2C, SPI)
        channel - either board pin number or BCM number depending on which mode is set.
        """
        board = getBoard()
        logger.info("GPIO function of Channel : {} is {}".format(channel,channel_config[channel].direction))
    
    
    class PWM:
        # initialise PWM channel
        def __init__(self, channel, frequency):
            """
            x.__init__(...) initializes x; see help(type(x)) for signature
            """
            self.chanel = channel
            self.frequency = frequency
            self.dutycycle = 0
            global channel_config
            channel_config[channel] = Channel(channel,PWM,)
            board = getBoard()
            logger.info("Initialized PWM for Channel : {} at frequency : {}".format(channel,frequency))
        
        # where dc is the duty cycle (0.0 <= dc <= 100.0)
        def start(self, dutycycle):
            """
            Start software PWM
            dutycycle - the duty cycle (0.0 to 100.0)
            """
            self.dutycycle = dutycycle
            board = getBoard()
            logger.info("start pwm on channel : {} with Duty cycle : {}".format(self.chanel,dutycycle))
        
        # where freq is the new frequency in Hz
        def ChangeFrequency(self, frequency):
            """
            Change the frequency
            frequency - frequency in Hz (freq > 1.0)
            """
            board = getBoard()
            logger.info("Freqency Changed for channel : {} from : {} -> to : {}".format(self.chanel,self.frequency,frequency))
            self.frequency = frequency
    
        # where 0.0 <= dc <= 100.0
        def ChangeDutyCycle(self, dutycycle):
            """
            Change the duty cycle
            dutycycle - between 0.0 and 100.0
            """
            board = getBoard()
            self.dutycycle = dutycycle
            logger.info("Dutycycle Changed for channel : {} from : {} -> to : {}".format(self.chanel,self.dutycycle,dutycycle))
        
        # stop PWM generation
        def stop(self):
            board = getBoard()
            logger.info("Stop pwm on channel : {} with Duty cycle : {}".format(self.chanel,self.dutycycle))
    
    
    def cleanup(channel=None):
        """
        Clean up by resetting all GPIO channels that have been used by this program to INPUT with no pullup/pulldown and no event detection
        [channel] - individual channel or list/tuple of channels to clean up.  Default - clean every channel that has been used.
        """
        board = getBoard()
        if channel is not None:
            logger.info("Cleaning Up Channel : {}".format(channel))
        else:
            logger.info("Cleaning Up all channels")
            
    def getBoard():
        rpib = PiBoard.Board.getInstance()
        if rpib == None:
            rpib = PiBoard.Board()
        return rpib
       
 

Как вы можете видеть, «единственное» отличие от оригинала заключается в том, что мой GPIO создает экземпляр PiBoard и получает эти моменты в методах. В методе add_event_detect , который я вызываю setChannelEvent в PiBoard.

Следующее решение является уродливым, но оно работает. Требуется некоторая логика, но у нее есть служба сокетов внутри потока, живущего внутри синглтона. Когда служба вызывается, в конечном итоге вызывается внешний метод ext_callback .

 import threading
import time
import socket as sk


class Board:
    channelConfigs = {}
    channelEvents = {}
    __instance = None
    serviceThread = None
    
    @staticmethod 
    def getInstance():
        """ Static access method. """
        if Board.__instance == None:
            Board()
        return Board.__instance
    
    def __init__(self):
        if Board.__instance != None:
            raise Exception("This class is a singleton!")
        else:
            Board.__instance = self
            Board.__instance.serviceThread = ServiceThread()
            Board.__instance.serviceThread.setPiBoardCallback(Board.__instance.piBoardCallback)
            Board.__instance.serviceThread.threadify()
           
    def piBoardCallback(_piBoardInstance, _value):
        
        global channelEvents
        # This assumes that _value is in format {channel:[HI|LOW]}, i.e. 22:HI
        values = _value.split(":")
        channel = values[0]
        edge = values[1]
        
        event = _piBoardInstance.channelEvents[int(channel)]
        # TODO: Handle logic on wether to call event callback or not.
        event.eventCallback(event)
    
    def setChannelConfig(_piBoardInstance, channel):
        if channel != None:
            _piBoardInstance.channelConfigs[channel.chanel] = channel
        
    def setChannelEvent(_piBoardInstance, _channel, _edge, _channelEventCallback):
        
        if _channel != None:
            event = Event(_edge, _channelEventCallback, _channel)
            _piBoardInstance.channelEvents[_channel] = event


class Event:

    eventCallback = None
    edge = None
    channel = None

    def __init__(self, _edge, _eventCallback, _channel):
        self.eventCallback = _eventCallback
        self.edge = _edge
        self.channel = _channel

   
class Service:
 
    serviceThreadCallback = None
 
    def __init__(self):
        print(self)
        
    def listen(self, _serviceThreadCallback):
        global serviceThreadCallback
        serviceThreadCallback = _serviceThreadCallback
        connection = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
        connection.setsockopt(sk.SOL_SOCKET, sk.SO_REUSEADDR, 1)
        connection.bind(('0.0.0.0', 5566))
        connection.listen(10)
        while True:
            current_connection, address = connection.accept()
            while True:
                data = current_connection.recv(2048)
    
                if data == 'quit\n':
                    current_connection.shutdown(1)
                    current_connection.close()
                    break
    
                elif data == 'stop\n':
                    current_connection.shutdown(1)
                    current_connection.close()
                    exit()
    
                elif data:
                    _serviceThreadCallback(data)
                    
                else:
                    break         
    
    def setCallback(_serviceThreadCallback):
        global serviceThreadCallback
        serviceThreadCallback = _serviceThreadCallback

                
class ServiceThread:

    thread = None
    svc = None
    piBoardCallback = None

    def __init__(self, interval=1):
        self.interval = interval

    def run(self):
            global piBoardCallback
            self.svc = Service()
            self.svc.listen(piBoardCallback)

    def setPiBoardCallback(_serviceThread, _piBoardCallback):
        global piBoardCallback
        piBoardCallback = _piBoardCallback
         
    def threadify(self):
        global thread
        thread = threading.Thread(target=self.run)
        thread.daemon = True  # Daemonize thread
        thread.start()  # Start the execution
        

def ext_callback(_event):
    print "ext_callback"
    print _event.channel       
    print _event.edge       
    print _event.eventCallback

def getBoard():
    _rpib = Board.getInstance()
    if _rpib == None:
        _rpib = Board()
    return _rpib

class Channel:
    def __init__(self,channel, direction, initial=0,pull_up_down=0):
        self.chanel = channel
        self.direction = direction
        self.initial = initial
        self.current = initial
        self.pull_up_down = pull_up_down

           
if __name__ == '__main__': 
     
    try:
        while True:
            rpib = getBoard()
            rpib.setChannelConfig(Channel(22, 32, 0, 0))
            rpib.setChannelEvent(22, 32, ext_callback)
            time.sleep(1000)
    except KeyboardInterrupt:
        pass