Initializes a new python-can based socket, described by the provided arguments and keyword arguments. This SocketWrapper gets automatically registered in the SocketsPool. :param args: Arguments for the python-can Bus object :param kwargs: Keyword arguments for the py
(self, *args, **kwargs)
| 201 | """Helper class to wrap a python-can Bus object as socket""" |
| 202 | |
| 203 | def __init__(self, *args, **kwargs): |
| 204 | # type: (Tuple[Any, ...], Dict[str, Any]) -> None |
| 205 | """Initializes a new python-can based socket, described by the provided |
| 206 | arguments and keyword arguments. This SocketWrapper gets automatically |
| 207 | registered in the SocketsPool. |
| 208 | |
| 209 | :param args: Arguments for the python-can Bus object |
| 210 | :param kwargs: Keyword arguments for the python-can Bus object |
| 211 | """ |
| 212 | super(SocketWrapper, self).__init__(*args, **kwargs) |
| 213 | self.lock = threading.Lock() |
| 214 | self.rx_queue = deque() # type: deque[can_Message] |
| 215 | self.name = None # type: Optional[str] |
| 216 | SocketsPool.register(self, *args, **kwargs) |
| 217 | |
| 218 | def _recv_internal(self, timeout): |
| 219 | # type: (int) -> Tuple[Optional[can_Message], bool] |