Sistema di eventi in Python


Answers

Lo sto facendo in questo modo:

class Event(list):
    """Event subscription.

    A list of callable objects. Calling an instance of this will cause a
    call to each item in the list in ascending order by index.

    Example Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> e = Event()
    >>> e()
    >>> e.append(f)
    >>> e(123)
    f(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    f(10)
    g(10)
    >>> del e[0]
    >>> e(2)
    g(2)

    """
    def __call__(self, *args, **kwargs):
        for f in self:
            f(*args, **kwargs)

    def __repr__(self):
        return "Event(%s)" % list.__repr__(self)

Comunque, come con tutto il resto che ho visto, non c'è pydoc generato automaticamente per questo, e niente firme, che fa davvero schifo.

Question

Quale sistema di eventi per Python usi? Sono già a conoscenza del pydispatcher , ma mi stavo chiedendo che altro si possa trovare o è comunemente usato?

Non sono interessato ai gestori di eventi che fanno parte di framework di grandi dimensioni, preferirei utilizzare una soluzione ridotta che possa essere facilmente estesa.




Se volevi fare cose più complicate, come unire gli eventi o riprovare, puoi usare il modello Osservabile e una libreria matura che implementa ciò. https://github.com/ReactiveX/RxPY . Gli osservabili sono molto comuni in Javascript e Java e sono molto comodi da usare per alcune attività asincrone.

from rx import Observable, Observer


def push_five_strings(observer):
        observer.on_next("Alpha")
        observer.on_next("Beta")
        observer.on_next("Gamma")
        observer.on_next("Delta")
        observer.on_next("Epsilon")
        observer.on_completed()


class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):
        print("Done!")

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.create(push_five_strings)

source.subscribe(PrintObserver())

OUTPUT :

Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon
Done!



Io uso zope.event . Sono le ossa più nude che puoi immaginare. :-) Infatti, ecco il codice sorgente completo:

subscribers = []

def notify(event):
    for subscriber in subscribers:
        subscriber(event)

Si noti che non è possibile inviare messaggi tra processi, ad esempio. Non è un sistema di messaggistica, solo un sistema di eventi, niente di più, niente di meno.




Ecco un design minimale che dovrebbe funzionare bene. Quello che devi fare è semplicemente ereditare Observer in una classe e successivamente usare observe(event_name, callback_fn) per ascoltare un evento specifico. Ogni volta che l'evento specifico viene attivato in qualsiasi parte del codice (ad esempio, Event('USB connected') ), il corrispondente richiamo si attiva.

class Observer():
    _observers = []
    def __init__(self):
        self._observers.append(self)
        self._observed_events = []
    def observe(self, event_name, callback_fn):
        self._observed_events.append({'event_name' : event_name, 'callback_fn' : callback_fn})


class Event():
    def __init__(self, event_name, *callback_args):
        for observer in Observer._observers:
            for observable in observer._observed_events:
                if observable['event_name'] == event_name:
                    observable['callback_fn'](*callback_args)

Esempio:

class Room(Observer):
    def __init__(self):
        print("Room is ready.")
        Observer.__init__(self) # DON'T FORGET THIS
    def someone_arrived(self, who):
        print(who + " has arrived!")

# Observe for specific event
room = Room()
room.observe('someone arrived',  room.someone_arrived)

# Fire some events
Event('someone left',    'John')
Event('someone arrived', 'Lenard') # will output "Lenard has arrived!"
Event('someone Farted',  'Lenard')



Ho creato una classe EventManager (codice alla fine). La sintassi è la seguente:

#Create an event with no listeners assigned to it
EventManager.addEvent( eventName = [] )

#Create an event with listeners assigned to it
EventManager.addEvent( eventName = [fun1, fun2,...] )

#Create any number event with listeners assigned to them
EventManager.addEvent( eventName1 = [e1fun1, e1fun2,...], eventName2 = [e2fun1, e2fun2,...], ... )

#Add or remove listener to an existing event
EventManager.eventName += extra_fun
EventManager.eventName -= removed_fun

#Delete an event
del EventManager.eventName

#Fire the event
EventManager.eventName()

Ecco un esempio:

def hello(name):
    print "Hello {}".format(name)

def greetings(name):
    print "Greetings {}".format(name)

EventManager.addEvent( salute = [greetings] )
EventManager.salute += hello

print "\nInitial salute"
EventManager.salute('Oscar')

print "\nNow remove greetings"
EventManager.salute -= greetings
EventManager.salute('Oscar')

Produzione:

Saluto iniziale
Saluti Oscar
Ciao Oscar

Ora rimuovi i saluti
Ciao Oscar

Codice EventManger:

class EventManager:

    class Event:
        def __init__(self,functions):
            if type(functions) is not list:
                raise ValueError("functions parameter has to be a list")
            self.functions = functions

        def __iadd__(self,func):
            self.functions.append(func)
            return self

        def __isub__(self,func):
            self.functions.remove(func)
            return self

        def __call__(self,*args,**kvargs):
            for func in self.functions : func(*args,**kvargs)

    @classmethod
    def addEvent(cls,**kvargs):
        """
        addEvent( event1 = [f1,f2,...], event2 = [g1,g2,...], ... )
        creates events using **kvargs to create any number of events. Each event recieves a list of functions,
        where every function in the list recieves the same parameters.

        Example:

        def hello(): print "Hello ",
        def world(): print "World"

        EventManager.addEvent( salute = [hello] )
        EventManager.salute += world

        EventManager.salute()

        Output:
        Hello World
        """
        for key in kvargs.keys():
            if type(kvargs[key]) is not list:
                raise ValueError("value has to be a list")
            else:
                kvargs[key] = cls.Event(kvargs[key])

        cls.__dict__.update(kvargs)



Potresti dare un'occhiata a pymitter ( pypi ). Si tratta di un piccolo approccio a file singolo (~ 250 loc) che fornisce "namespace, caratteri jolly e TTL".

Ecco un esempio di base:

from pymitter import EventEmitter

ee = EventEmitter()

# decorator usage
@ee.on("myevent")
def handler1(arg):
   print "handler1 called with", arg

# callback usage
def handler2(arg):
    print "handler2 called with", arg
ee.on("myotherevent", handler2)

# emit
ee.emit("myevent", "foo")
# -> "handler1 called with foo"

ee.emit("myotherevent", "bar")
# -> "handler2 called with bar"