Registers a SocketWrapper object. Every SocketWrapper describes to a python-can bus object. This python-can bus object can only exist once. In case this object already exists in this SocketsPool, organized by a SocketMapper object, the new SocketWrapper is inserted in the
(self, socket, *args, **kwargs)
| 138 | self.last_call = time.monotonic() |
| 139 | |
| 140 | def register(self, socket, *args, **kwargs): |
| 141 | # type: (SocketWrapper, Tuple[Any, ...], Dict[str, Any]) -> None |
| 142 | """Registers a SocketWrapper object. Every SocketWrapper describes to |
| 143 | a python-can bus object. This python-can bus object can only exist |
| 144 | once. In case this object already exists in this SocketsPool, organized |
| 145 | by a SocketMapper object, the new SocketWrapper is inserted in the |
| 146 | list of subscribers of the SocketMapper. Otherwise a new python-can |
| 147 | Bus object is created from the provided args and kwargs and inserted, |
| 148 | encapsulated in a SocketMapper, into this SocketsPool. |
| 149 | |
| 150 | :param socket: SocketWrapper object which needs to be registered. |
| 151 | :param args: Arguments for the python-can Bus object |
| 152 | :param kwargs: Keyword arguments for the python-can Bus object |
| 153 | """ |
| 154 | if "interface" in kwargs.keys(): |
| 155 | k = str(kwargs.get("interface", "unknown_interface")) + "_" + \ |
| 156 | str(kwargs.get("channel", "unknown_channel")) |
| 157 | else: |
| 158 | k = str(kwargs.get("bustype", "unknown_bustype")) + "_" + \ |
| 159 | str(kwargs.get("channel", "unknown_channel")) |
| 160 | with self.pool_mutex: |
| 161 | if k in self.pool: |
| 162 | t = self.pool[k] |
| 163 | t.sockets.append(socket) |
| 164 | filters = [s.filters for s in t.sockets |
| 165 | if s.filters is not None] |
| 166 | if filters: |
| 167 | t.bus.set_filters(reduce(add, filters)) |
| 168 | socket.name = k |
| 169 | else: |
| 170 | bus = can_Bus(*args, **kwargs) |
| 171 | socket.name = k |
| 172 | self.pool[k] = SocketMapper(bus, [socket]) |
| 173 | |
| 174 | def unregister(self, socket): |
| 175 | # type: (SocketWrapper) -> None |
no test coverage detected