(self)
| 342 | self._init_socket() |
| 343 | |
| 344 | def _init_socket(self) -> None: |
| 345 | can_socket = socket.socket( |
| 346 | socket.PF_CAN, socket.SOCK_DGRAM, CAN_ISOTP) |
| 347 | |
| 348 | self.__set_option_flags( |
| 349 | can_socket, |
| 350 | self.ext_address, |
| 351 | self.rx_ext_address, |
| 352 | self.listen_only, |
| 353 | self.padding, |
| 354 | self.frame_txtime) |
| 355 | |
| 356 | can_socket.setsockopt( |
| 357 | SOL_CAN_ISOTP, |
| 358 | CAN_ISOTP_RECV_FC, |
| 359 | self.__build_can_isotp_fc_options(stmin=self.stmin, bs=self.bs)) |
| 360 | |
| 361 | tx_flags = ((CANFD_FDF if self.fd else 0) + |
| 362 | (CANFD_BRS if (self.brs + self.fd) else 0)) |
| 363 | tx_dl = CAN_FD_ISOTP_DEFAULT_LL_TX_DL if self.fd else CAN_ISOTP_DEFAULT_LL_TX_DL |
| 364 | |
| 365 | can_socket.setsockopt( |
| 366 | SOL_CAN_ISOTP, |
| 367 | CAN_ISOTP_LL_OPTS, |
| 368 | self.__build_can_isotp_ll_options( |
| 369 | mtu=CAN_ISOTP_CANFD_MTU if self.fd else CAN_ISOTP_DEFAULT_LL_MTU, |
| 370 | tx_dl=tx_dl, |
| 371 | tx_flags=tx_flags)) |
| 372 | can_socket.setsockopt( |
| 373 | socket.SOL_SOCKET, |
| 374 | SO_TIMESTAMPNS, |
| 375 | 1 |
| 376 | ) |
| 377 | |
| 378 | self.__bind_socket(can_socket, self.iface, self.tx_id, self.rx_id) |
| 379 | # make sure existing sockets are closed, |
| 380 | # required in case of a reconnect. |
| 381 | if getattr(self, "outs", None): |
| 382 | if getattr(self, "ins", None) != self.outs: |
| 383 | if self.outs and self.outs.fileno() != -1: |
| 384 | self.outs.close() |
| 385 | if getattr(self, "ins", None): |
| 386 | if self.ins.fileno() != -1: |
| 387 | self.ins.close() |
| 388 | |
| 389 | self.ins = can_socket |
| 390 | self.outs = can_socket |
| 391 | self.closed = False |
| 392 | |
| 393 | def recv_raw(self, x=0xffff): |
| 394 | # type: (int) -> Tuple[Optional[Type[Packet]], Optional[bytes], Optional[float]] # noqa: E501 |
no test coverage detected