44This module contains the implementation of :class:`~can.Notifier`.
55"""
66
7+ from typing import Iterable , List , Optional , Union
8+
9+ from can .bus import BusABC
10+ from can .listener import Listener
11+ from can .message import Message
12+
713import threading
814import logging
915import time
1319
1420
1521class Notifier :
16- def __init__ (self , bus , listeners , timeout = 1.0 , loop = None ):
22+ def __init__ (
23+ self ,
24+ bus : BusABC ,
25+ listeners : Iterable [Listener ],
26+ timeout : float = 1.0 ,
27+ loop : Optional [asyncio .AbstractEventLoop ] = None ,
28+ ):
1729 """Manages the distribution of :class:`can.Message` instances to listeners.
1830
1931 Supports multiple buses and listeners.
@@ -24,37 +36,40 @@ def __init__(self, bus, listeners, timeout=1.0, loop=None):
2436 many listeners carry out flush operations to persist data.
2537
2638
27- :param can.BusABC bus: A :ref:`bus` or a list of buses to listen to.
28- :param list listeners: An iterable of :class:`~can.Listener`
29- :param float timeout: An optional maximum number of seconds to wait for any message.
30- :param asyncio.AbstractEventLoop loop:
31- An :mod:`asyncio` event loop to schedule listeners in.
39+ :param bus: A :ref:`bus` or a list of buses to listen to.
40+ :param listeners: An iterable of :class:`~can.Listener`
41+ :param timeout: An optional maximum number of seconds to wait for any message.
42+ :param loop: An :mod:`asyncio` event loop to schedule listeners in.
3243 """
33- self .listeners = listeners
44+ self .listeners = list ( listeners )
3445 self .bus = bus
3546 self .timeout = timeout
3647 self ._loop = loop
3748
3849 #: Exception raised in thread
39- self .exception = None
50+ self .exception : Optional [ Exception ] = None
4051
4152 self ._running = True
4253 self ._lock = threading .Lock ()
4354
44- self ._readers = []
55+ self ._readers : List [ Union [ int , threading . Thread ]] = []
4556 buses = self .bus if isinstance (self .bus , list ) else [self .bus ]
4657 for bus in buses :
4758 self .add_bus (bus )
4859
49- def add_bus (self , bus ):
60+ def add_bus (self , bus : BusABC ):
5061 """Add a bus for notification.
5162
52- :param can.BusABC bus:
63+ :param bus:
5364 CAN bus instance.
5465 """
55- if self ._loop is not None and hasattr (bus , "fileno" ) and bus .fileno () >= 0 :
66+ if (
67+ self ._loop is not None
68+ and hasattr (bus , "fileno" )
69+ and bus .fileno () >= 0 # type: ignore
70+ ):
5671 # Use file descriptor to watch for messages
57- reader = bus .fileno ()
72+ reader = bus .fileno () # type: ignore
5873 self ._loop .add_reader (reader , self ._on_message_available , bus )
5974 else :
6075 reader = threading .Thread (
@@ -66,11 +81,11 @@ def add_bus(self, bus):
6681 reader .start ()
6782 self ._readers .append (reader )
6883
69- def stop (self , timeout = 5 ):
84+ def stop (self , timeout : float = 5 ):
7085 """Stop notifying Listeners when new :class:`~can.Message` objects arrive
7186 and call :meth:`~can.Listener.stop` on each Listener.
7287
73- :param float timeout:
88+ :param timeout:
7489 Max time in seconds to wait for receive threads to finish.
7590 Should be longer than timeout given at instantiation.
7691 """
@@ -81,14 +96,14 @@ def stop(self, timeout=5):
8196 now = time .time ()
8297 if now < end_time :
8398 reader .join (end_time - now )
84- else :
99+ elif self . _loop :
85100 # reader is a file descriptor
86101 self ._loop .remove_reader (reader )
87102 for listener in self .listeners :
88103 if hasattr (listener , "stop" ):
89104 listener .stop ()
90105
91- def _rx_thread (self , bus ):
106+ def _rx_thread (self , bus : BusABC ):
92107 msg = None
93108 try :
94109 while self ._running :
@@ -109,40 +124,38 @@ def _rx_thread(self, bus):
109124 self ._on_error (exc )
110125 raise
111126
112- def _on_message_available (self , bus ):
127+ def _on_message_available (self , bus : BusABC ):
113128 msg = bus .recv (0 )
114129 if msg is not None :
115130 self ._on_message_received (msg )
116131
117- def _on_message_received (self , msg ):
132+ def _on_message_received (self , msg : Message ):
118133 for callback in self .listeners :
119134 res = callback (msg )
120135 if self ._loop is not None and asyncio .iscoroutine (res ):
121136 # Schedule coroutine
122137 self ._loop .create_task (res )
123138
124- def _on_error (self , exc ):
139+ def _on_error (self , exc : Exception ):
125140 for listener in self .listeners :
126141 if hasattr (listener , "on_error" ):
127142 listener .on_error (exc )
128143
129- def add_listener (self , listener ):
144+ def add_listener (self , listener : Listener ):
130145 """Add new Listener to the notification list.
131146 If it is already present, it will be called two times
132147 each time a message arrives.
133148
134- :param can.Listener listener: Listener to be added to
135- the list to be notified
149+ :param listener: Listener to be added to the list to be notified
136150 """
137151 self .listeners .append (listener )
138152
139- def remove_listener (self , listener ):
153+ def remove_listener (self , listener : Listener ):
140154 """Remove a listener from the notification list. This method
141155 trows an exception if the given listener is not part of the
142156 stored listeners.
143157
144- :param can.Listener listener: Listener to be removed from
145- the list to be notified
158+ :param listener: Listener to be removed from the list to be notified
146159 :raises ValueError: if `listener` was never added to this notifier
147160 """
148161 self .listeners .remove (listener )
0 commit comments