Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def command(self, op=None, data=b"", chk=0, wait_response=True, timeout=DEFAULT_TIMEOUT):
saved_timeout = self._port.timeout
new_timeout = min(timeout, MAX_TIMEOUT)
if new_timeout != saved_timeout:
self._port.timeout = new_timeout
try:
if op is not None:
self.trace("command op=0x%02x data len=%s wait_response=%d timeout=%.3f data=%s",
op, len(data), 1 if wait_response else 0, timeout, HexFormatter(data))
pkt = struct.pack(b'
read_bytes = port.read(1 if waiting == 0 else waiting)
if read_bytes == b'':
waiting_for = "header" if partial_packet is None else "content"
trace_function("Timed out waiting for packet %s", waiting_for)
raise FatalError("Timed out waiting for packet %s" % waiting_for)
trace_function("Read %d bytes: %s", len(read_bytes), HexFormatter(read_bytes))
for b in read_bytes:
if type(b) is int:
b = bytes([b]) # python 2/3 compat
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
partial_packet = None
else: # normal byte in packet
partial_packet += b
"""Generator to read SLIP packets from a serial port.
Yields one full SLIP packet at a time, raises exception on timeout or invalid data.
Designed to avoid too many calls to serial.read(1), which can bog
down on slow systems.
"""
partial_packet = None
in_escape = False
while True:
waiting = port.inWaiting()
read_bytes = port.read(1 if waiting == 0 else waiting)
if read_bytes == b'':
waiting_for = "header" if partial_packet is None else "content"
trace_function("Timed out waiting for packet %s", waiting_for)
raise FatalError("Timed out waiting for packet %s" % waiting_for)
trace_function("Read %d bytes: %s", len(read_bytes), HexFormatter(read_bytes))
for b in read_bytes:
if type(b) is int:
b = bytes([b]) # python 2/3 compat
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
read_bytes = port.read(1 if waiting == 0 else waiting)
if read_bytes == b'':
waiting_for = "header" if partial_packet is None else "content"
trace_function("Timed out waiting for packet %s", waiting_for)
raise FatalError("Timed out waiting for packet %s" % waiting_for)
trace_function("Read %d bytes: %s", len(read_bytes), HexFormatter(read_bytes))
for b in read_bytes:
if type(b) is int:
b = bytes([b]) # python 2/3 compat
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
partial_packet = None
else: # normal byte in packet
partial_packet += b
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
partial_packet = None
else: # normal byte in packet
partial_packet += b
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
partial_packet += b'\xdb'
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid SLIP escape (0xdb, 0x%s)' % (hexify(b)))
elif b == b'\xdb': # start of escape sequence
in_escape = True
elif b == b'\xc0': # end of packet
trace_function("Received full packet: %s", HexFormatter(partial_packet))
yield partial_packet
partial_packet = None
else: # normal byte in packet
partial_packet += b
"""Generator to read SLIP packets from a serial port.
Yields one full SLIP packet at a time, raises exception on timeout or invalid data.
Designed to avoid too many calls to serial.read(1), which can bog
down on slow systems.
"""
partial_packet = None
in_escape = False
while True:
waiting = port.inWaiting()
read_bytes = port.read(1 if waiting == 0 else waiting)
if read_bytes == b'':
waiting_for = "header" if partial_packet is None else "content"
trace_function("Timed out waiting for packet %s", waiting_for)
raise FatalError("Timed out waiting for packet %s" % waiting_for)
trace_function("Read %d bytes: %s", len(read_bytes), HexFormatter(read_bytes))
for b in read_bytes:
if type(b) is int:
b = bytes([b]) # python 2/3 compat
if partial_packet is None: # waiting for packet header
if b == b'\xc0':
partial_packet = b""
else:
trace_function("Read invalid data: %s", HexFormatter(read_bytes))
trace_function("Remaining data in serial buffer: %s", HexFormatter(port.read(port.inWaiting())))
raise FatalError('Invalid head of packet (0x%s)' % hexify(b))
elif in_escape: # part-way through escape sequence
in_escape = False
if b == b'\xdc':
partial_packet += b'\xc0'
elif b == b'\xdd':
def write(self, packet):
buf = b'\xc0' \
+ (packet.replace(b'\xdb',b'\xdb\xdd').replace(b'\xc0',b'\xdb\xdc')) \
+ b'\xc0'
self.trace("Write %d bytes: %s", len(buf), HexFormatter(buf))
self._port.write(buf)