4908056526

Build Type

push

github

Committed by Andreas Wicenec

Commit Message

Fixed small issues with existing graphs

Run Details

20 of 25 new or added lines in 2 files covered. (80.0%)

88 existing lines in 5 files now uncovered.

15342 of 19053 relevant lines covered (80.52%)

1.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.5

/daliuge-engine/dlg/event.py

#
#    ICRAR - International Centre for Radio Astronomy Research
#    (c) UWA - The University of Western Australia, 2014
#    Copyright by UWA (in the framework of the ICRAR)
#    All rights reserved
#
#    This library is free software; you can redistribute it and/or
#    modify it under the terms of the GNU Lesser General Public
#    License as published by the Free Software Foundation; either
#    version 2.1 of the License, or (at your option) any later version.
#
#    This library 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
#    Lesser General Public License for more details.
#
#    You should have received a copy of the GNU Lesser General Public
#    License along with this library; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
#    MA 02111-1307  USA
#

from collections import defaultdict
import logging
from abc import ABC, abstractmethod
from typing import Optional, Union

logger = logging.getLogger(__name__)


class Event(object):
    """
    An event sent through the DALiuGE framework.

    Events have at least a field describing the type of event they are (instead
    of having subclasses of the `Event` class), and therefore this class makes
    sure that at least that field exists. Any other piece of information can be
    attached to individual instances of this class, depending on the event type.
    """

    def __init__(self, type: str):
        self.type = type

    def __repr__(self, *args, **kwargs):
        return "<Event %r>" % (self.__dict__)


class EventHandler(ABC):
    @abstractmethod
    def handleEvent(self, e: Event) -> None:
        pass


class EventFirer(object):
    """
    An object that fires events.

    Objects that have an interest on receiving events from this object subscribe
    to it via the `subscribe` method; likewise they can unsubscribe from it via
    the `unsubscribe` method. Events are handled to the listeners by calling
    their `handleEvent` method with the event as its sole argument.

    Listeners can specify the type of event they listen to at subscription time,
    or can also prefer to receive all events fired by this object if they wish
    so.
    """

    __ALL_EVENTS = object()

    def __init__(self):
        # Union string key with object to handle __ALL_EVENTS above
        self._listeners: defaultdict[
            Union[str, object], list[EventHandler]
        ] = defaultdict(list)

    def subscribe(
        self, listener: EventHandler, eventType: Optional[str] = None
    ):
        """
        Subscribes `listener` to events fired by this object. If `eventType` is
        not `None` then `listener` will only receive events of `eventType` that
        originate from this object, otherwise it will receive all events.
        """
        # logger.debug(
        #     "Adding listener to %r eventType=%s: %r",
        #     self,
        #     eventType,
        #     listener,
        # )
        eventType = eventType or EventFirer.__ALL_EVENTS
        self._listeners[eventType].append(listener)

    def unsubscribe(
        self, listener: EventHandler, eventType: Optional[str] = None
    ):
        """
        Unsubscribes `listener` from events fired by this object.
        """
        logger.debug(
            "Removing listener to %r eventType=%s: %r",
            self.oid,
            eventType,
            listener.oid,
        ) if hasattr(listener, "oid") else None

        eventType = eventType or EventFirer.__ALL_EVENTS
        if listener in self._listeners[eventType]:
            self._listeners[eventType].remove(listener)

    def _fireEvent(self, eventType: str, **attrs):
        """
        Delivers an event of `eventType` to all interested listeners.

        All the key-value pairs contained in `attrs` are set as attributes of
        the event being sent.
        """

        # Which listeners should we call?
        listeners: list[EventHandler] = []
        if eventType in self._listeners:
            listeners += self._listeners[eventType]
        if EventFirer.__ALL_EVENTS in self._listeners:
            listeners += self._listeners[EventFirer.__ALL_EVENTS]
        if not listeners:
            logger.debug("No listeners found for eventType=%s", eventType)
            return

        # Now that we are sure there are listeners for our event
        # create it and send it to all of them
        e = Event(eventType)

        for k, v in attrs.items():
            setattr(e, k, v)

        for l in listeners:
            l.handleEvent(e)

1	#
2	# ICRAR - International Centre for Radio Astronomy Research
3	# (c) UWA - The University of Western Australia, 2014
4	# Copyright by UWA (in the framework of the ICRAR)
5	# All rights reserved
6	#
7	# This library is free software; you can redistribute it and/or
8	# modify it under the terms of the GNU Lesser General Public
9	# License as published by the Free Software Foundation; either
10	# version 2.1 of the License, or (at your option) any later version.
11	#
12	# This library is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15	# Lesser General Public License for more details.
16	#
17	# You should have received a copy of the GNU Lesser General Public
18	# License along with this library; if not, write to the Free Software
19	# Foundation, Inc., 59 Temple Place, Suite 330, Boston,
20	# MA 02111-1307 USA
21	#
22
23	from collections import defaultdict	2✔
24	import logging	2✔
25	from abc import ABC, abstractmethod	2✔
26	from typing import Optional, Union	2✔
27
28	logger = logging.getLogger(__name__)	2✔
29
30
31	class Event(object):	2✔
32	"""
33	An event sent through the DALiuGE framework.
34
35	Events have at least a field describing the type of event they are (instead
36	of having subclasses of the `Event` class), and therefore this class makes
37	sure that at least that field exists. Any other piece of information can be
38	attached to individual instances of this class, depending on the event type.
39	"""
40
41	def __init__(self, type: str):	2✔
42	self.type = type	2✔
43
44	def __repr__(self, args, *kwargs):	2✔
45	return "<Event %r>" % (self.__dict__)	×
46
47
48	class EventHandler(ABC):	2✔
49	@abstractmethod	2✔
50	def handleEvent(self, e: Event) -> None:	2✔
51	pass	×
52
53
54	class EventFirer(object):	2✔
55	"""
56	An object that fires events.
57
58	Objects that have an interest on receiving events from this object subscribe
59	to it via the `subscribe` method; likewise they can unsubscribe from it via
60	the `unsubscribe` method. Events are handled to the listeners by calling
61	their `handleEvent` method with the event as its sole argument.
62
63	Listeners can specify the type of event they listen to at subscription time,
64	or can also prefer to receive all events fired by this object if they wish
65	so.
66	"""
67
68	__ALL_EVENTS = object()	2✔
69
70	def __init__(self):	2✔
71	# Union string key with object to handle __ALL_EVENTS above
72	self._listeners: defaultdict[	2✔
73	Union[str, object], list[EventHandler]
74	] = defaultdict(list)
75
76	def subscribe(	2✔
77	self, listener: EventHandler, eventType: Optional[str] = None
78	):
79	"""
80	Subscribes `listener` to events fired by this object. If `eventType` is
81	not `None` then `listener` will only receive events of `eventType` that
82	originate from this object, otherwise it will receive all events.
83	"""
84	# logger.debug(
85	# "Adding listener to %r eventType=%s: %r",
86	# self,
87	# eventType,
88	# listener,
89	# )
90	eventType = eventType or EventFirer.__ALL_EVENTS	2✔
91	self._listeners[eventType].append(listener)	2✔
92
93	def unsubscribe(	2✔
94	self, listener: EventHandler, eventType: Optional[str] = None
95	):
96	"""
97	Unsubscribes `listener` from events fired by this object.
98	"""
99	logger.debug(	2✔
100	"Removing listener to %r eventType=%s: %r",
101	self.oid,
102	eventType,
103	listener.oid,
104	) if hasattr(listener, "oid") else None
105
106	eventType = eventType or EventFirer.__ALL_EVENTS	2✔
107	if listener in self._listeners[eventType]:	2✔
UNCOV 108	self._listeners[eventType].remove(listener)	×
109
110	def _fireEvent(self, eventType: str, **attrs):	2✔
111	"""
112	Delivers an event of `eventType` to all interested listeners.
113
114	All the key-value pairs contained in `attrs` are set as attributes of
115	the event being sent.
116	"""
117
118	# Which listeners should we call?
119	listeners: list[EventHandler] = []	2✔
120	if eventType in self._listeners:	2✔
121	listeners += self._listeners[eventType]	2✔
122	if EventFirer.__ALL_EVENTS in self._listeners:	2✔
123	listeners += self._listeners[EventFirer.__ALL_EVENTS]	2✔
124	if not listeners:	2✔
125	logger.debug("No listeners found for eventType=%s", eventType)	2✔
126	return	2✔
127
128	# Now that we are sure there are listeners for our event
129	# create it and send it to all of them
130	e = Event(eventType)	2✔
131
132	for k, v in attrs.items():	2✔
133	setattr(e, k, v)	2✔
134
135	for l in listeners:	2✔
136	l.handleEvent(e)	2✔

ICRAR / daliuge / 4908056526

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous