From 64ba08e7aee9afb533c2998442d776e0b16bfd07 Mon Sep 17 00:00:00 2001 From: iabdalkader Date: Sun, 19 Feb 2023 18:27:44 +0100 Subject: [PATCH] scripts/libraries: Format RPC/RTSP libraries. --- scripts/libraries/rpc.py | 709 ++++++++++++++++++++++++-------------- scripts/libraries/rtsp.py | 221 ++++++++---- 2 files changed, 605 insertions(+), 325 deletions(-) diff --git a/scripts/libraries/rpc.py b/scripts/libraries/rpc.py index 67fde9977..feb7a883e 100644 --- a/scripts/libraries/rpc.py +++ b/scripts/libraries/rpc.py @@ -7,24 +7,25 @@ import gc, network, omv, pyb, select, socket, stm, struct -class rpc: +class rpc: _COMMAND_HEADER_PACKET_MAGIC = 0x1209 _COMMAND_DATA_PACKET_MAGIC = 0xABD1 _RESULT_HEADER_PACKET_MAGIC = 0x9021 _RESULT_DATA_PACKET_MAGIC = 0x1DBA @micropython.viper - def __def_crc_16(self, data, size : int) -> int: # private + def __def_crc_16(self, data, size: int) -> int: # private d = ptr8(data) crc = 0xFFFF for i in range(size): crc ^= d[i] << 8 - for j in range(8): crc = (crc << 1) ^ (0x1021 if crc & 0x8000 else 0) + for j in range(8): + crc = (crc << 1) ^ (0x1021 if crc & 0x8000 else 0) return crc & 0xFFFF @micropython.viper - def __stm_crc_16(self, data, size : int) -> int: # private + def __stm_crc_16(self, data, size: int) -> int: # private ptr32(stm.CRC + stm.CRC_CR)[0] = (1 << 3) | 1 crc8 = ptr8(stm.CRC + stm.CRC_DR) d = ptr8(data) @@ -33,31 +34,34 @@ class rpc: return ptr32(stm.CRC + stm.CRC_DR)[0] @micropython.viper - def _zero(self, buff, size : int): # private + def _zero(self, buff, size: int): # private d = ptr8(buff) - for i in range(size): d[i] = 0 + for i in range(size): + d[i] = 0 @micropython.viper - def _same(self, data, size : int) -> bool: # private - if not size: return False + def _same(self, data, size: int) -> bool: # private + if not size: + return False d = ptr8(data) old = d[0] for i in range(1, size): new = d[i] - if new != old: return False + if new != old: + return False old = new return True # djb2 algorithm; see http://www.cse.yorku.ca/~oz/hash.html @micropython.viper - def _hash(self, data, size : int) -> uint: # private + def _hash(self, data, size: int) -> uint: # private h = 5381 d = ptr8(data) for i in range(size): h = ((h << 5) + h) ^ d[i] return uint(h) - def __init__(self): # private + def __init__(self): # private self.__crc_16 = self.__def_crc_16 if omv.board_type() == "H7": stm.mem32[stm.RCC + stm.RCC_AHB4ENR] = stm.mem32[stm.RCC + stm.RCC_AHB4ENR] | (1 << 19) @@ -73,7 +77,7 @@ class rpc: buff = bytearray(payload_len + 4) return (buff, memoryview(buff)[2:-2]) - def _get_packet(self, magic_value, payload_buf_tuple, timeout): # private + def _get_packet(self, magic_value, payload_buf_tuple, timeout): # private packet = self.get_bytes(payload_buf_tuple[0], timeout) if packet is not None: magic = packet[0] | (packet[1] << 8) @@ -82,71 +86,91 @@ class rpc: return payload_buf_tuple[1] return None - def _set_packet(self, magic_value, payload=bytes()): # private + def _set_packet(self, magic_value, payload=bytes()): # private new_payload = bytearray(len(payload) + 4) new_payload[:2] = struct.pack("> 1) ^ (0xB8 if tx_lfsr & 1 else 0x00) - def stream_writer(self, call_back, write_timeout_ms=5000): # public + def stream_writer(self, call_back, write_timeout_ms=5000): # public packet = self._stream_get_bytes(bytearray(8), 1000) - if packet is None: return + if packet is None: + return magic = packet[0] | (packet[1] << 8) crc = packet[-2] | (packet[-1] << 8) - if magic != 0xEDF6 and crc != self.__crc_16(packet, len(packet) - 2): return - queue_depth = max(min(struct.unpack("> 1) ^ (0xB8 if rx_lfsr & 1 else 0x00) credits += 1 if credits > 0: data = call_back() - try: self._stream_put_bytes(self._set_packet(0x542E, struct.pack("= self._get_short_timeout: return None + while self.__pin.value(): + if pyb.elapsed_millis(start) >= self._get_short_timeout: + return None self.__spi.init(pyb.SPI.SLAVE, polarity=self.__polarity, phase=self.__clk_phase) - try: self.__spi.send_recv(buff, buff, timeout=timeout_ms) # SPI.recv() is broken. - except OSError: buff = None + try: + self.__spi.send_recv(buff, buff, timeout=timeout_ms) # SPI.recv() is broken. + except OSError: + buff = None self.__spi.deinit() return buff - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected start = pyb.millis() - while(self.__pin.value()): - if pyb.elapsed_millis(start) >= self._put_short_timeout: return + while self.__pin.value(): + if pyb.elapsed_millis(start) >= self._put_short_timeout: + return self.__spi.init(pyb.SPI.SLAVE, polarity=self.__polarity, phase=self.__clk_phase) - try: self.__spi.send(data, timeout=timeout_ms) - except OSError: pass + try: + self.__spi.send(data, timeout=timeout_ms) + except OSError: + pass self.__spi.deinit() -class rpc_uart_master(rpc_master): - def __init__(self, baudrate=9600, uart_port=3): # private +class rpc_uart_master(rpc_master): + def __init__(self, baudrate=9600, uart_port=3): # private self.__uart = pyb.UART(uart_port, baudrate, timeout=2, timeout_char=2) rpc_master.__init__(self) - def _flush(self): # protected + def _flush(self): # protected self.__uart.read(self.__uart.any()) - def get_bytes(self, buff, timeout_ms): # protected + def get_bytes(self, buff, timeout_ms): # protected if self.__uart.readinto(buff) is None: pyb.delay(self._get_short_timeout) return None return buff - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected self.__uart.write(data) - def _stream_get_bytes(self, buff, timeout_ms): # protected + def _stream_get_bytes(self, buff, timeout_ms): # protected p = select.poll() p.register(self.__uart, select.POLLIN) p.poll(1000) return self.get_bytes(buff, timeout_ms) -class rpc_uart_slave(rpc_slave): - def __init__(self, baudrate=9600, uart_port=3): # private +class rpc_uart_slave(rpc_slave): + def __init__(self, baudrate=9600, uart_port=3): # private self.__uart = pyb.UART(uart_port, baudrate, timeout=2, timeout_char=2) rpc_slave.__init__(self) - def _flush(self): # protected + def _flush(self): # protected self.__uart.read(self.__uart.any()) - def get_bytes(self, buff, timeout_ms): # protected - if self.__uart.readinto(buff) is None: return None + def get_bytes(self, buff, timeout_ms): # protected + if self.__uart.readinto(buff) is None: + return None return buff - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected self.__uart.write(data) - def _stream_get_bytes(self, buff, timeout_ms): # protected + def _stream_get_bytes(self, buff, timeout_ms): # protected p = select.poll() p.register(self.__uart, select.POLLIN) p.poll(1000) return self.get_bytes(buff, timeout_ms) -class rpc_usb_vcp_master(rpc_master): - def __init__(self): # private +class rpc_usb_vcp_master(rpc_master): + def __init__(self): # private self.__usb_vcp = pyb.USB_VCP() - if self.__usb_vcp.debug_mode_enabled(): raise OSError("You cannot use the USB VCP while the IDE is connected!") + if self.__usb_vcp.debug_mode_enabled(): + raise OSError("You cannot use the USB VCP while the IDE is connected!") self.__usb_vcp.setinterrupt(-1) rpc_master.__init__(self) - def _flush(self): # protected + def _flush(self): # protected self.__usb_vcp.read() - def get_bytes(self, buff, timeout_ms): # protected - if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): return None + def get_bytes(self, buff, timeout_ms): # protected + if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): + return None return buff - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected self.__usb_vcp.send(data, timeout=timeout_ms) -class rpc_usb_vcp_slave(rpc_slave): - def __init__(self): # private +class rpc_usb_vcp_slave(rpc_slave): + def __init__(self): # private self.__usb_vcp = pyb.USB_VCP() - if self.__usb_vcp.debug_mode_enabled(): raise OSError("You cannot use the USB VCP while the IDE is connected!") + if self.__usb_vcp.debug_mode_enabled(): + raise OSError("You cannot use the USB VCP while the IDE is connected!") self.__usb_vcp.setinterrupt(-1) rpc_slave.__init__(self) - def _flush(self): # protected + def _flush(self): # protected self.__usb_vcp.read() - def get_bytes(self, buff, timeout_ms): # protected - if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): return None + def get_bytes(self, buff, timeout_ms): # protected + if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): + return None return buff - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected self.__usb_vcp.send(data, timeout=timeout_ms) -class rpc_network_master(rpc_master): - def __valid_tcp_socket(self): # private +class rpc_network_master(rpc_master): + def __valid_tcp_socket(self): # private if self.__tcp__socket is None: try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -557,26 +700,28 @@ class rpc_network_master(rpc_master): s.settimeout(1) self.__tcp__socket, addr = s.accept() s.close() - except OSError: self.__tcp__socket = None + except OSError: + self.__tcp__socket = None return self.__tcp__socket is not None - def __close_tcp_socket(self): # private + def __close_tcp_socket(self): # private self.__tcp__socket.close() self.__tcp__socket = None - def __valid_udp_socket(self): # private + def __valid_udp_socket(self): # private if self.__udp__socket is None: try: self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.__udp__socket.bind(self.__myaddr) - except OSError: self.__udp__socket = None + except OSError: + self.__udp__socket = None return self.__udp__socket is not None - def __close_udp_socket(self): # private + def __close_udp_socket(self): # private self.__udp__socket.close() self.__udp__socket = None - def __init__(self, network_if, port=0x1DBA): # private + def __init__(self, network_if, port=0x1DBA): # private self._udp_limit = 1400 self._timeout_scale = 10 self.__network = network_if @@ -588,78 +733,99 @@ class rpc_network_master(rpc_master): print("IP Address:Port %s:%d\nRunning..." % self.__myaddr) rpc_master.__init__(self) - def _flush(self): # protected + def _flush(self): # protected if self.__valid_udp_socket(): try: self.__udp__socket.settimeout(0.001) - while(True): + while True: data, addr = self.__udp__socket.recvfrom(1400) - if not len(data): break - except OSError: self.__close_udp_socket() + if not len(data): + break + except OSError: + self.__close_udp_socket() if self.__tcp__socket is not None: try: self.__tcp__socket.settimeout(0.001) - while(True): + while True: data = self.__tcp__socket.recv(1400) - if not len(data): break - except OSError: self.__close_tcp_socket() + if not len(data): + break + except OSError: + self.__close_tcp_socket() - def get_bytes(self, buff, timeout_ms): # protected + def get_bytes(self, buff, timeout_ms): # protected i = 0 l = len(buff) if l <= self._udp_limit: if self.__valid_udp_socket(): try: - self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale) + self.__udp__socket.settimeout( + self._get_short_timeout * 0.001 * self._timeout_scale + ) while l: data, addr = self.__udp__socket.recvfrom(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data i += data_len l -= data_len # We don't need to close the socket on error since it's connectionless. - except OSError: self.__close_udp_socket() + except OSError: + self.__close_udp_socket() elif self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: data = self.__tcp__socket.recv(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() return buff if not l else None - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected i = 0 l = len(data) if l <= self._udp_limit: if self.__valid_udp_socket(): try: - self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale) + self.__udp__socket.settimeout( + self._put_short_timeout * 0.001 * self._timeout_scale + ) while l: - data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__slave_addr) - if not data_len: break + data_len = self.__udp__socket.sendto( + data[i : i + min(l, 1400)], self.__slave_addr + ) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_udp_socket() - except OSError: self.__close_udp_socket() + if l: + self.__close_udp_socket() + except OSError: + self.__close_udp_socket() elif self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: - data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)]) - if not data_len: break + data_len = self.__tcp__socket.send(data[i : i + min(l, 1400)]) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() - def _stream_get_bytes(self, buff, timeout_ms): # protected + def _stream_get_bytes(self, buff, timeout_ms): # protected i = 0 l = len(buff) if self.__valid_tcp_socket(): @@ -668,56 +834,65 @@ class rpc_network_master(rpc_master): while l: data = self.__tcp__socket.recv(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() return buff if not l else None - def _stream_put_bytes(self, data, timeout_ms): # protected + def _stream_put_bytes(self, data, timeout_ms): # protected i = 0 l = len(data) if self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: - data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)]) - if not data_len: break + data_len = self.__tcp__socket.send(data[i : i + min(l, 1400)]) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() - if l: raise OSError # Stop Stream. + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() + if l: + raise OSError # Stop Stream. + class rpc_network_slave(rpc_slave): - - def __valid_tcp_socket(self): # private + def __valid_tcp_socket(self): # private if self.__tcp__socket is None: try: self.__tcp__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.__tcp__socket.connect(self.__master_addr) - except OSError: self.__tcp__socket = None + except OSError: + self.__tcp__socket = None return self.__tcp__socket is not None - def __close_tcp_socket(self): # private + def __close_tcp_socket(self): # private self.__tcp__socket.close() self.__tcp__socket = None - def __valid_udp_socket(self): # private + def __valid_udp_socket(self): # private if self.__udp__socket is None: try: self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.__udp__socket.bind(self.__myaddr) - except OSError: self.__udp__socket = None + except OSError: + self.__udp__socket = None return self.__udp__socket is not None - def __close_udp_socket(self): # private + def __close_udp_socket(self): # private self.__udp__socket.close() self.__udp__socket = None - def __init__(self, network_if, port=0x1DBA): # private + def __init__(self, network_if, port=0x1DBA): # private self._udp_limit = 1400 self._timeout_scale = 10 self.__network = network_if @@ -729,79 +904,100 @@ class rpc_network_slave(rpc_slave): print("IP Address:Port %s:%d\nRunning..." % self.__myaddr) rpc_slave.__init__(self) - def _flush(self): # protected + def _flush(self): # protected if self.__valid_udp_socket(): try: self.__udp__socket.settimeout(0.001) - while(True): + while True: data, addr = self.__udp__socket.recvfrom(1400) - if not len(data): break - except OSError: self.__close_udp_socket() + if not len(data): + break + except OSError: + self.__close_udp_socket() if self.__tcp__socket is not None: try: self.__tcp__socket.settimeout(0.001) - while(True): + while True: data = self.__tcp__socket.recv(1400) - if not len(data): break - except OSError: self.__close_tcp_socket() + if not len(data): + break + except OSError: + self.__close_tcp_socket() - def get_bytes(self, buff, timeout_ms): # protected + def get_bytes(self, buff, timeout_ms): # protected i = 0 l = len(buff) if l <= self._udp_limit: if self.__valid_udp_socket(): try: - self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale) + self.__udp__socket.settimeout( + self._get_short_timeout * 0.001 * self._timeout_scale + ) while l: data, addr = self.__udp__socket.recvfrom(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data self.__master_addr = addr i += data_len l -= data_len # We don't need to close the socket on error since it's connectionless. - except OSError: self.__close_udp_socket() + except OSError: + self.__close_udp_socket() elif self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: data = self.__tcp__socket.recv(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() return buff if not l else None - def put_bytes(self, data, timeout_ms): # protected + def put_bytes(self, data, timeout_ms): # protected i = 0 l = len(data) if l <= self._udp_limit: if self.__valid_udp_socket(): try: - self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale) + self.__udp__socket.settimeout( + self._put_short_timeout * 0.001 * self._timeout_scale + ) while l: - data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__master_addr) - if not data_len: break + data_len = self.__udp__socket.sendto( + data[i : i + min(l, 1400)], self.__master_addr + ) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_udp_socket() - except OSError: self.__close_udp_socket() + if l: + self.__close_udp_socket() + except OSError: + self.__close_udp_socket() elif self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: - data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)]) - if not data_len: break + data_len = self.__tcp__socket.send(data[i : i + min(l, 1400)]) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() - def _stream_get_bytes(self, buff, timeout_ms): # protected + def _stream_get_bytes(self, buff, timeout_ms): # protected i = 0 l = len(buff) if self.__valid_tcp_socket(): @@ -810,25 +1006,32 @@ class rpc_network_slave(rpc_slave): while l: data = self.__tcp__socket.recv(min(l, 1400)) data_len = len(data) - if not data_len: break - buff[i:i+data_len] = data + if not data_len: + break + buff[i : i + data_len] = data i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() return buff if not l else None - def _stream_put_bytes(self, data, timeout_ms): # protected + def _stream_put_bytes(self, data, timeout_ms): # protected i = 0 l = len(data) if self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(timeout_ms * 0.001) while l: - data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)]) - if not data_len: break + data_len = self.__tcp__socket.send(data[i : i + min(l, 1400)]) + if not data_len: + break i += data_len l -= data_len - if l: self.__close_tcp_socket() - except OSError: self.__close_tcp_socket() - if l: raise OSError # Stop Stream. + if l: + self.__close_tcp_socket() + except OSError: + self.__close_tcp_socket() + if l: + raise OSError # Stop Stream. diff --git a/scripts/libraries/rtsp.py b/scripts/libraries/rtsp.py index 5a0b9d81c..19e5a2ba5 100644 --- a/scripts/libraries/rtsp.py +++ b/scripts/libraries/rtsp.py @@ -7,9 +7,9 @@ import network, pyb, re, socket, struct -class rtsp_server: - def __valid_tcp_socket(self): # private +class rtsp_server: + def __valid_tcp_socket(self): # private if self.__tcp__socket is None: try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -18,26 +18,28 @@ class rtsp_server: s.settimeout(5) self.__tcp__socket, self.__client_addr = s.accept() s.close() - except OSError: self.__tcp__socket = None + except OSError: + self.__tcp__socket = None return self.__tcp__socket is not None - def __close_tcp_socket(self): # private + def __close_tcp_socket(self): # private self.__tcp__socket.close() self.__tcp__socket = None - def __valid_udp_socket(self): # private + def __valid_udp_socket(self): # private if self.__udp__socket is None: try: self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.__udp__socket.bind((self.__myip, self.__client_rtp_port)) - except OSError: self.__udp__socket = None + except OSError: + self.__udp__socket = None return self.__udp__socket is not None - def __close_udp_socket(self): # private + def __close_udp_socket(self): # private self.__udp__socket.close() self.__udp__socket = None - def __init__(self, network_if, port=554): # private + def __init__(self, network_if, port=554): # private self.__network = network_if self.__myip = self.__network.ifconfig()[0] self.__myaddr = (self.__myip, port) @@ -56,100 +58,148 @@ class rtsp_server: self.__ssrc = pyb.rng() print("IP Address:Port %s:%d\nRunning..." % self.__myaddr) - def register_setup_cb(self, cb): # public + def register_setup_cb(self, cb): # public self.__setup_cb = cb - def register_play_cb(self, cb): # public + def register_play_cb(self, cb): # public self.__play_cb = cb - def register_pause_cb(self, cb): # public + def register_pause_cb(self, cb): # public self.__pause_cb = cb - def register_teardown_cb(self, cb): # public + def register_teardown_cb(self, cb): # public self.__teardown_cb = cb - def __send_rtsp_response(self, code, name, extra=""): # private + def __send_rtsp_response(self, code, name, extra=""): # private self.__tcp__socket.send("RTSP/1.0 %d %s\r\n%s\r\n" % (code, name, extra)) - def __send_rtsp_response_cseq(self, code, name, seq, extra=""): # private + def __send_rtsp_response_cseq(self, code, name, seq, extra=""): # private self.__send_rtsp_response(code, name, "CSeq: %d\r\n%s" % (seq, extra)) - def __send_rtsp_response_ok(self, seq, extra=""): # private + def __send_rtsp_response_ok(self, seq, extra=""): # private self.__send_rtsp_response_cseq(200, "OK", seq, extra) - def __parse_rtsp_request(self, data): # private - if data[0] == 0x24: return # Interleaved RTCP Packet - if data[0] >= 0x80: return # Ressambled TCP - s = list(filter(lambda x: x and not x.startswith("User-Agent"), data.decode().splitlines())) + def __parse_rtsp_request(self, data): # private + if data[0] == 0x24: + return # Interleaved RTCP Packet + if data[0] >= 0x80: + return # Ressambled TCP + s = list( + filter(lambda x: x and not x.startswith("User-Agent"), data.decode().splitlines()) + ) if s and len(s) >= 2: - line0 = s[0].split(' ') + line0 = s[0].split(" ") request = line0[0] - self.__pathname = re.sub("rtsp(u)?://[a-zA-Z0-9\-\.]+(:\d+)?(/)?", '/', line0[1]) - if self.__pathname != '/' and self.__pathname.endswith('/'): + self.__pathname = re.sub("rtsp(u)?://[a-zA-Z0-9\-\.]+(:\d+)?(/)?", "/", line0[1]) + if self.__pathname != "/" and self.__pathname.endswith("/"): self.__pathname = self.__pathname[:-1] if line0[2] == "RTSP/1.0": - if len(s) >= 3 and s[1].split(' ')[0] != "CSeq:": + if len(s) >= 3 and s[1].split(" ")[0] != "CSeq:": for i in range(2, len(s)): - if s[i].split(' ')[0] == "CSeq:": + if s[i].split(" ")[0] == "CSeq:": temp = s[i] s[i] = s[1] s[1] = temp break - line1 = s[1].split(' ') + line1 = s[1].split(" ") if line1[0] == "CSeq:": seq = int(line1[1]) if request == "OPTIONS": if len(s) >= 3: m = re.match("(Proxy-)?Require: (\S)+", s[2]) if m: - self.__send_rtsp_response_cseq(551, "Option not supported", seq, "Unsupported: %s\r\n" % m.group(2)) + self.__send_rtsp_response_cseq( + 551, + "Option not supported", + seq, + "Unsupported: %s\r\n" % m.group(2), + ) return - self.__send_rtsp_response_ok(seq, "Public: DESCRIBE, SETUP, PLAY, PAUSE, TEARDOWN\r\n") + self.__send_rtsp_response_ok( + seq, "Public: DESCRIBE, SETUP, PLAY, PAUSE, TEARDOWN\r\n" + ) return elif request == "DESCRIBE": - payload = "v=0\r\no=- %d %d IN IP4 %s\r\ns=OpenMV Video\r\nt=0 0\r\nm=video 0 RTP/AVP 26" % (pyb.rng(), pyb.rng(), self.__myip) - self.__send_rtsp_response_ok(seq, "Content-Type: application/sdp\r\nContent-Length: %d\r\n\r\n%s" % (len(payload) + 2, payload)) + payload = ( + "v=0\r\no=- %d %d IN IP4 %s\r\ns=OpenMV Video\r\nt=0 0\r\nm=video 0 RTP/AVP 26" + % (pyb.rng(), pyb.rng(), self.__myip) + ) + self.__send_rtsp_response_ok( + seq, + "Content-Type: application/sdp\r\nContent-Length: %d\r\n\r\n%s" + % (len(payload) + 2, payload), + ) return elif request == "SETUP": if len(s) >= 3: - m = re.match("Transport: RTP/AVP(/UDP)?;unicast;client_port=(\d+)-(\d+)", s[2]) + m = re.match( + "Transport: RTP/AVP(/UDP)?;unicast;client_port=(\d+)-(\d+)", s[2] + ) if m: self.__transport_is_tcp = False self.__client_rtp_port = int(m.group(2)) self.__client_rtcp_port = int(m.group(3)) - self.__client_rtp_addr = (self.__client_addr[0], self.__client_rtp_port) - self.__client_rtcp_addr = (self.__client_addr[0], self.__client_rtcp_port) + self.__client_rtp_addr = ( + self.__client_addr[0], + self.__client_rtp_port, + ) + self.__client_rtcp_addr = ( + self.__client_addr[0], + self.__client_rtcp_port, + ) self.__session = pyb.rng() self.__ssrc = pyb.rng() - self.__send_rtsp_response_ok(seq, "%s;server_port=%d-%d;ssrc=%d\r\nSession: %d\r\n" % (s[2], self.__client_rtp_port, self.__client_rtcp_port, self.__ssrc, self.__session)) - if self.__setup_cb: self.__setup_cb(self.__pathname, self.__session) + self.__send_rtsp_response_ok( + seq, + "%s;server_port=%d-%d;ssrc=%d\r\nSession: %d\r\n" + % ( + s[2], + self.__client_rtp_port, + self.__client_rtcp_port, + self.__ssrc, + self.__session, + ), + ) + if self.__setup_cb: + self.__setup_cb(self.__pathname, self.__session) return - m = re.match("Transport: RTP/AVP/TCP;unicast;interleaved=(\d+)-(\d+)", s[2]) + m = re.match( + "Transport: RTP/AVP/TCP;unicast;interleaved=(\d+)-(\d+)", s[2] + ) if m: self.__transport_is_tcp = True self.__client_rtp_channel = int(m.group(1)) self.__client_rtcp_channel = int(m.group(2)) self.__session = pyb.rng() self.__ssrc = pyb.rng() - self.__send_rtsp_response_ok(seq, "%s\r\nSession: %d\r\n" % (s[2], self.__session)) - if self.__setup_cb: self.__setup_cb(self.__pathname, self.__session) + self.__send_rtsp_response_ok( + seq, "%s\r\nSession: %d\r\n" % (s[2], self.__session) + ) + if self.__setup_cb: + self.__setup_cb(self.__pathname, self.__session) return elif request == "PLAY": - s = list(filter(lambda x: x and not x.startswith("Range"), s)) - if len(s) >= 3: + s = list(filter(lambda x: x and not x.startswith("Range"), s)) + if len(s) >= 3: m = re.match("Session: (\d+)", s[2]) if m: self.__playing = True - self.__send_rtsp_response_ok(seq, "RTP-Info: url=%s;seq=%d\r\n" % (line0[1], self.__sequence_number)) - if self.__play_cb: self.__play_cb(self.__pathname, int(m.group(1))) + self.__send_rtsp_response_ok( + seq, + "RTP-Info: url=%s;seq=%d\r\n" + % (line0[1], self.__sequence_number), + ) + if self.__play_cb: + self.__play_cb(self.__pathname, int(m.group(1))) return elif request == "PAUSE": - if len(s) >= 3: + if len(s) >= 3: m = re.match("Session: (\d+)", s[2]) if m: self.__playing = False self.__send_rtsp_response_ok(seq) - if self.__pause_cb: self.__pause_cb(self.__pathname, int(m.group(1))) + if self.__pause_cb: + self.__pause_cb(self.__pathname, int(m.group(1))) return elif request == "TEARDOWN": if len(s) >= 3: @@ -157,65 +207,92 @@ class rtsp_server: if m: self.__playing = False self.__send_rtsp_response_ok(seq) - if self.__teardown_cb: self.__teardown_cb(self.__pathname, int(m.group(1))) + if self.__teardown_cb: + self.__teardown_cb(self.__pathname, int(m.group(1))) return else: self.__send_rtsp_response_cseq(501, "Not Implemented", seq) return self.__send_rtsp_response(400, "Bad Request") - def __valid_socket(self): # private - if self.__transport_is_tcp: return self.__client_rtp_channel is not None - else: return self.__valid_udp_socket() and self.__client_rtp_addr is not None + def __valid_socket(self): # private + if self.__transport_is_tcp: + return self.__client_rtp_channel is not None + else: + return self.__valid_udp_socket() and self.__client_rtp_addr is not None - def __settimeout(self, timeout): # private - if self.__transport_is_tcp: self.__tcp__socket.settimeout(timeout) - else: self.__udp__socket.settimeout(timeout) + def __settimeout(self, timeout): # private + if self.__transport_is_tcp: + self.__tcp__socket.settimeout(timeout) + else: + self.__udp__socket.settimeout(timeout) - def __send(self, data): # private - if self.__transport_is_tcp: return self.__tcp__socket.send(struct.pack(">BBH", 0x24, self.__client_rtp_channel, len(data)) + data) - else: return self.__udp__socket.sendto(data, self.__client_rtp_addr) + def __send(self, data): # private + if self.__transport_is_tcp: + return self.__tcp__socket.send( + struct.pack(">BBH", 0x24, self.__client_rtp_channel, len(data)) + data + ) + else: + return self.__udp__socket.sendto(data, self.__client_rtp_addr) - def __close_socket(self): # private - if self.__transport_is_tcp: pass - else: self.__close_udp_socket() + def __close_socket(self): # private + if self.__transport_is_tcp: + pass + else: + self.__close_udp_socket() - def __send_rtp(self, image_callback, quality): # private + def __send_rtp(self, image_callback, quality): # private img = image_callback(self.__pathname, self.__session).compress(quality) - if img.width() >= 2040: raise ValueError("Maximum width is 2040") - if img.height() >= 2040: raise ValueError("Maximum height is 2040") + if img.width() >= 2040: + raise ValueError("Maximum width is 2040") + if img.height() >= 2040: + raise ValueError("Maximum height is 2040") i = 0 l = img.size() rtp_header_size = 12 jpeg_header_size = 8 max_packet_size = 1400 - rtp_header_size - jpeg_header_size - if self.__transport_is_tcp: max_packet_size -= 4 + if self.__transport_is_tcp: + max_packet_size -= 4 if self.__valid_socket(): try: self.__settimeout(0.1) timestamp = (pyb.millis() * 90) & 0xFFFFFFFF while l: - rtp_header = struct.pack(">BBHII", - 0x80, 0x9A if l <= max_packet_size else 0x1A, - self.__sequence_number, timestamp, self.__ssrc) + rtp_header = struct.pack( + ">BBHII", + 0x80, + 0x9A if l <= max_packet_size else 0x1A, + self.__sequence_number, + timestamp, + self.__ssrc, + ) self.__sequence_number = (self.__sequence_number + 1) & 0xFFFF - jpeg_header = struct.pack(">IBBBB", i, 0, quality, int(img.width() // 8), int(img.height() // 8)) - img_data = img.bytearray()[i:i+min(l, max_packet_size)] + jpeg_header = struct.pack( + ">IBBBB", i, 0, quality, int(img.width() // 8), int(img.height() // 8) + ) + img_data = img.bytearray()[i : i + min(l, max_packet_size)] img_data_len = len(img_data) data_len = self.__send(rtp_header + jpeg_header + img_data) - if not data_len: break + if not data_len: + break i += img_data_len l -= img_data_len - if l: self.__close_socket() - except OSError: self.__close_socket() + if l: + self.__close_socket() + except OSError: + self.__close_socket() - def stream(self, image_callback, quality=90): # public + def stream(self, image_callback, quality=90): # public while True: if self.__valid_tcp_socket(): try: self.__tcp__socket.settimeout(0.1) while True: data = self.__tcp__socket.recv(1400) - if data and len(data): self.__parse_rtsp_request(data) - if self.__playing: self.__send_rtp(image_callback, quality) - except OSError: self.__close_tcp_socket() + if data and len(data): + self.__parse_rtsp_request(data) + if self.__playing: + self.__send_rtp(image_callback, quality) + except OSError: + self.__close_tcp_socket()