Add generic serial tracing for debug

This causes us to wrap all serial object usage in a module that will
trace (hex and ascii) communication in a generic way. These traces
will be stored (max 10 per session) separately, submitted as part of
a debug report, and purged at exit.

Note that this also makes us gzip any files over 1MB, which is the
limit for attachment sizes on the website.
This commit is contained in:
Dan Smith 2025-08-14 19:36:12 -07:00 committed by Dan Smith
parent da7b1863cf
commit e773e04e6e
6 changed files with 247 additions and 9 deletions

View file

@ -14,6 +14,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import datetime
import gzip
import logging
import os
import platform
@ -32,6 +33,7 @@ from chirp import logger
from chirp import platform as chirp_platform
from chirp.wxui import common
from chirp.wxui import config
from chirp.wxui import serialtrace
_ = wx.GetTranslation
CONF = config.get()
@ -126,6 +128,15 @@ def prepare_report(chirpmain):
manifest['files']['debug_log.txt'] = f.read()
tmpf = tempfile.mktemp('.config', 'chirp')
# Grab any trace files
for tracefile in serialtrace.TRACEFILES:
if os.path.exists(tracefile):
LOG.debug('Capturing serial trace file %s', tracefile)
with open(tracefile, 'rb') as f:
manifest['files'][os.path.basename(tracefile)] = f.read()
else:
LOG.debug('Serial trace file %s does not exist', tracefile)
return manifest
@ -631,16 +642,28 @@ class ResultPage(BugReportPage):
if 'issue' not in manifest:
self._create_bug(manifest)
for fn in list(manifest['files'].keys()):
fdata = manifest['files'][fn]
if len(fdata) > 1024 * 1024:
LOG.warning('File %s is larger than 1MB, compressing', fn)
fdata = gzip.compress(fdata)
manifest['files'].pop(fn)
fn += '.gz'
manifest['files'][fn] = fdata
tokens = []
for fn in manifest['files']:
token = self._upload_file(manifest, fn)
if fn.lower().endswith('.img'):
ct = 'application/octet-stream'
else:
ext = os.path.splitext(fn)[1].lower()
if ext in ('.log', '.txt'):
ct = 'text/plain'
tokens.append({'token': token,
'filename': fn,
'content_type': ct})
else:
ct = 'application/octet-stream'
token_info = {'token': token,
'filename': fn,
'content_type': ct}
tokens.append(token_info)
LOG.debug('File tokens: %s', tokens)
notes = '[Uploaded from CHIRP %s]\n\n' % CHIRP_VERSION

View file

@ -33,6 +33,7 @@ from chirp import errors
from chirp.wxui import config
from chirp.wxui import common
from chirp.wxui import developer
from chirp.wxui import serialtrace
_ = wx.GetTranslation
LOG = logging.getLogger(__name__)
@ -163,8 +164,9 @@ def open_serial(port, rclass):
pipe.open()
pipe.baudrate = rclass.BAUD_RATE
else:
pipe = serial.Serial(baudrate=rclass.BAUD_RATE,
rtscts=rclass.HARDWARE_FLOW, timeout=0.25)
pipe = serialtrace.SerialTrace(
baudrate=rclass.BAUD_RATE,
rtscts=rclass.HARDWARE_FLOW, timeout=0.25)
pipe.rts = rclass.WANTS_RTS
pipe.dtr = rclass.WANTS_DTR
pipe.port = port

View file

@ -51,6 +51,7 @@ from chirp.wxui import query_sources
from chirp.wxui import radioinfo
from chirp.wxui import radiothread
from chirp.wxui import report
from chirp.wxui import serialtrace
from chirp.wxui import settingsedit
from chirp import CHIRP_VERSION
@ -1322,6 +1323,9 @@ class ChirpMain(wx.Frame):
'state')
config._CONFIG.save()
# Clean up any trace files we left
serialtrace.purge_trace_files(0)
ALL_MAIN_WINDOWS.remove(self)
self.Destroy()

129
chirp/wxui/serialtrace.py Normal file
View file

@ -0,0 +1,129 @@
# Copyright 2025 Dan Smith <chirp@f.danplanet.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import datetime
import logging
import os
import serial
import tempfile
import time
from chirp import util
from chirp.wxui import config
CONF = config.get()
LOG = logging.getLogger(__name__)
TRACEFILES = []
def get_trace_entry(direction, start_ts, data):
loglines = util.hexprint(data, block_size=16).split('\n')
ts = time.monotonic() - start_ts
loglines = ['%7.3f %s %s%s' % (ts, direction, line, os.linesep)
for line in loglines if line.strip()]
if not loglines and direction == 'R' and not data:
# No data read means timeout, so denote that for clarity
loglines = ['%7.3f %s # timeout%s' % (ts, direction, os.linesep)]
return loglines
def purge_trace_files(keep=10):
global TRACEFILES
if keep == 0:
purge = TRACEFILES
TRACEFILES = []
else:
purge = TRACEFILES[:-keep]
TRACEFILES = TRACEFILES[-10:]
for fn in purge:
try:
os.remove(fn)
LOG.debug('Removed old trace file %s', fn)
except FileNotFoundError:
pass
except Exception as e:
LOG.error('Failed to remove old trace file %s: %s', fn, e)
class SerialTrace(serial.Serial):
def __init__(self, *a, **k):
self.__tracef = None
super().__init__(*a, **k)
def open(self):
super().open()
try:
self.__trace_start = time.monotonic()
self.__tracef = tempfile.NamedTemporaryFile(mode='w',
delete=False,
prefix='chirp-trace-',
suffix='.txt')
TRACEFILES.append(self.__tracef.name)
purge_trace_files(10)
now = datetime.datetime.now()
self.log('Serial trace %s started at %s' % (self, now.isoformat()))
LOG.info('Serial trace file created: %s' % self.__tracef.name)
except Exception as e:
LOG.error('Failed to create serial trace file: %s' % e)
self.__tracef = None
def write(self, data):
super().write(data)
if self.__tracef:
try:
self.__tracef.writelines(get_trace_entry('W',
self.__trace_start,
data))
except Exception as e:
LOG.error('Failed to write to serial trace file: %s' % e)
self.__tracef = None
def read(self, size=1):
data = super().read(size)
if self.__tracef:
try:
self.__tracef.writelines(get_trace_entry('R',
self.__trace_start,
data))
except Exception as e:
LOG.error('Failed to write to serial trace file: %s' % e)
self.__tracef = None
return data
def close(self):
super().close()
if self.__tracef:
try:
now = datetime.datetime.now()
self.log('Trace ended at %s' % now.isoformat())
self.__tracef.close()
LOG.info('Serial trace file closed: %s' % self.__tracef.name)
except Exception as e:
LOG.error('Failed to close serial trace file: %s' % e)
finally:
self.__tracef = None
def log(self, message):
"""Log a message to the trace file.
Use this to annotate important events in the trace file, such as
reading a new block, etc.
"""
if self.__tracef:
try:
self.__tracef.write('# %s\n' % message)
except Exception as e:
LOG.error('Failed to write log message to trace file: %s' % e)
self.__tracef = None

View file

@ -5,6 +5,7 @@ from unittest import mock
from chirp import chirp_common
from chirp import errors
from chirp.wxui import serialtrace
from tests import base
LOG = logging.getLogger(__name__)
@ -14,7 +15,7 @@ class SerialException(Exception):
pass
class SerialNone:
class SerialNone(serialtrace.SerialTrace):
def flush(self):
pass

View file

@ -0,0 +1,79 @@
import unittest
from unittest import mock
from chirp.wxui import serialtrace
class TestSerialTrace(unittest.TestCase):
@mock.patch('serial.Serial.open')
def test_open(self, mock_open):
trace = serialtrace.SerialTrace()
self.assertIsNone(trace._SerialTrace__tracef)
trace.open()
self.assertIsNotNone(trace._SerialTrace__tracef)
self.assertTrue(trace._SerialTrace__tracef.name.endswith('.txt'))
mock_open.assert_called_once()
@mock.patch('os.remove')
def test_purge_trace_files(self, mock_remove):
from chirp.wxui import serialtrace
for i in range(15):
serialtrace.TRACEFILES.append('test_trace_%i.txt' % i)
files = serialtrace.TRACEFILES[:]
# Purge to 10 files keeps the last 10
serialtrace.purge_trace_files(10)
self.assertEqual(len(serialtrace.TRACEFILES), 10)
self.assertEqual(['test_trace_%i.txt' % (i + 5) for i in range(10)],
serialtrace.TRACEFILES)
# Purge to 20 does not change anything since only 10 stored
serialtrace.purge_trace_files(20)
self.assertEqual(10, len(serialtrace.TRACEFILES))
# Purge to zero removes all files
serialtrace.purge_trace_files(0)
self.assertEqual(0, len(serialtrace.TRACEFILES))
# Make sure we ended up removing all the files
mock_remove.assert_has_calls([mock.call(fn) for fn in files],
any_order=True)
@mock.patch('serial.Serial.open')
@mock.patch('serial.Serial.write')
@mock.patch('serial.Serial.read')
def test_log_write(self, mock_read, mock_write, mock_open):
mock_read.side_effect = [b'123', b'']
trace = serialtrace.SerialTrace()
trace.open()
fn = serialtrace.TRACEFILES[-1]
trace.write(b'foo')
trace.read(3)
trace.read(5)
trace.close()
with open(fn, 'r') as f:
content = f.read()
self.assertIn('# Serial trace', content)
self.assertIn('foo...', content)
self.assertIn('R # timeout', content)
@mock.patch('tempfile.NamedTemporaryFile')
@mock.patch('serial.Serial.open')
@mock.patch('serial.Serial.write')
@mock.patch('serial.Serial.read')
def test_log_write_fail(self, mock_read, mock_write, mock_open, mock_tf):
mock_tf.return_value.writelines.side_effect = [
None, Exception("Write error")]
trace = serialtrace.SerialTrace()
trace.open()
# This should generate a write failure
trace.read(3)
# Make sure we don't interrupt further communication
trace.write(b'foo')
trace.read()
trace.write(b'bar')
# Before we are closed, the trace file should have been abandoned
self.assertIsNone(trace._SerialTrace__tracef)