How to dismiss incomple received lines? #10
Comments
|
OK, unfortunately, changing the timeout didn't resolve the issue: I'm not sure what this means... |
|
Hi, I'm glad that you find the script useful :). I'll need to have a look at the source code. IIRC there is some functionality to skip the first line, which might be incomplete, but not afterwards. I wonder what might be a good way to detect such incomplete lines. Perhaps I could add a flag that when no newline is received within N seconds, the line is discarded, WDYT? |
Yes, that makes sense! |
|
I don't know how to add a flag, so I hard-coded a 10-second wait into the readloop like this: def ReadLoop(args, queue: persistent_queue.Queue):
"""Reads samples and stores them in a queue. Retries on IO errors."""
serial_fn: Optional[Callable[[BinaryIO], Generator[bytes, None,
None]]] = None
if args.serial_function:
module, fn_name = args.serial_function.rsplit(".", 1)
serial_fn = getattr(importlib.import_module(module), fn_name)
try:
logging.debug("Read loop started")
with serial.serial_for_url(args.device,
baudrate=args.baud_rate,
timeout=args.read_timeout) as handle:
if serial_fn:
lines = serial_fn(handle)
else:
lines = serial_samples.SerialLines(handle, args.max_line_length)
start_time = time.time()
for line in lines:
# If line does not end with newline character and more than 10 seconds have passed since its reception began
if not line.endswith(b'\n') and (time.time() - start_time > 10):
continue
# Reset the timer when we receive a complete transmission.
elif line.endswith(b'\n'):
start_time = time.time()
try:
line = str(line, encoding="UTF-8")
except TypeError:
pass
# Parse 'line', either with or without timestamp.
words = line.strip().split(" ")
if len(words) == 2:
(tags, values) = words
timestamp = int(time.time() * 1000000000)
elif len(words) == 3:
(tags, values, timestamp) = words
float(timestamp)
else:
raise ValueError("Unable to parse line {0!r}".format(line))
tags: str = ",".join(t for t in (tags, args.tags) if t)
queue.put("{0} {1} {2:d}".format(tags, values, timestamp))
except:
logging.exception("Error, retrying with backoff")
raiseI'm currently testing this and will let you now if it works (or if you spot a flaw right away, let me know). |
|
Upon further reflection, I realize that my test code is not actually discarding the incomplete line. But I don't understand whether that is a problem or not - it probably is - or how I can discard it. I don't understand how the serial data is read (i.e. how and when |
|
Not sure if this is any smarter than my initial try, but now I modified the def SerialLines(handle: BinaryIO, max_line_length: int) -> Generator[bytes, None, None]:
"""A generator that yields lines from a configured serial line.
Will never exit normally, only with an exception when there is an error
in the serial communication.
"""
SkipUntilNewLine(handle)
start_time = time.time() # initiates the start_time
buffer = b''
while True:
line: bytes = handle.readline(max_line_length)
logging.debug("Received line %r", line)
buffer += line
if buffer.endswith(b"\n"):
full_line = buffer
buffer = b'' # reset the buffer
start_time = time.time() # reset the time
yield full_line.rstrip()
elif time.time() - start_time > 10:
buffer = b'' # discard the buffer if no newline is received in 10 secs
start_time = time.time() # reset the time |
|
It didn't work, so I dove a bit more into the code and started wondering: shouldn't we actually be checking for It looks like the reason why the existing checks for |
|
FWIW, I'm trying this now: def check_incomplete_line(handle: BinaryIO, timeout: float, max_line_length: int):
'''Attempt to read a line within the given timeout.
Returns an incomplete line if timeout expires.'''
end_time = time.time() + timeout
line = b""
while time.time() < end_time and len(line) < max_line_length:
if handle.readable():
char = handle.read(1) # read one byte
line += char
if char == b'\n':
return line
return line # return incomplete line
def SerialLines(handle: BinaryIO, max_line_length: int, timeout: float) -> Generator[bytes, None, None]:
"""A generator that yields lines from a configured serial line.
Will never exit normally, only with an exception when there is an error
in the serial communication.
"""
SkipUntilNewLine(handle)
while True:
# replaced the handle.readline() call with our custom function
line: bytes = check_incomplete_line(handle, timeout, max_line_length)
logging.debug("Received line %r", line)
if not line.endswith(b"\n"):
raise LineOverflowError(line, max_line_length)
yield line.rstrip()Which means I'm calling This is too difficult for me. I obviously don't know what I'm doing. |
|
I think the problem in the previous version was that the timer started before anything was received. It looks like this works: def check_incomplete_line(handle: BinaryIO, timeout: float, max_line_length: int):
'''Attempt to read a line within the given timeout starting from the first received byte.
Returns nothing if timeout expires.'''
line = b""
end_time = None
while len(line) < max_line_length:
if handle.readable():
char = handle.read(1) # read one byte
if end_time is None: # if this is the first received byte
end_time = time.time() + timeout # set the end time
line += char
if char == b'\n':
return line
elif (time.time() >= end_time):
logging.warning("Dismissing incomplete line: %r", line)
return None # don't return the incomplete line
def SerialLines(handle: BinaryIO, max_line_length: int, timeout: float) -> Generator[bytes, None, None]:
"""A generator that yields lines from a configured serial line.
Will never exit normally, only with an exception when there is an error
in the serial communication.
"""
SkipUntilNewLine(handle)
while True:
# replaced the handle.readline() call with our custom function
line: bytes = check_incomplete_line(handle, timeout, max_line_length)
logging.debug("Received line %r", line)
if not line.endswith(b"\n"):
raise LineOverflowError(line, max_line_length)
yield line.rstrip()Here, the timer is started first when the first byte is received (i.e. that's when we set the For the sake of completeness: in order for this to work, we need to specify the timeout when the function is called in I have hard-coded 10 seconds, because figuring out how to add a flag that receives that value and adding code for when no timeout is specified would probably take me more hours of trial and error... I'm not even sure whether no flag should use the original code or some default timeout. The latter would be easier to implement. |
|
Thank you for looking into this! I'll try to incorporate your fix soon. |
I have been using your script for a few years now and it did me great service. Thanks for creating it.
Recently I had to change my setup and I can't seem to get the script running continuously on the new machine. At first, it runs fine, but after a few minutes up to a few hours, it gets stuck with this error:
I have included the last succeeding line for comparison.
If you take a closer look at the second (i.e. the failing) received line, you can see that it is cut off mid-way:
And after that, a new line is received, starting with
FTX...So, for some reason, the transmission of that line failed and we shall for now not care why. What I'm trying to understand is whether there is a way how the script can be made to handle (i.e. dismiss) such incomplete lines.
Is there a way?
As I write, it occurs to me that I might have set the timeout too high (I set it to 130 seconds, while the arduino is sending every 60 seconds), so if the line is dismissed if no end of line is received after the timeout, then that will save the problem. Will try it now. But let me know if there is another/better way.
The text was updated successfully, but these errors were encountered: