dominic/st
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

set_record.py: Add verbose tracing option

Browse Source 
The commented out print statements tracking values arriving
are now switchable using the --trace flag.
This commit is contained in:
Dominic Höglinger 2025-05-15 20:01:11 +02:00
parent 0c7e730438
commit bf27293543
1 changed files with 22 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
set_record.py
View File

@ -340,7 +340,7 @@ class Retagger:
self.packets_dropped += 1 self.packets_dropped += 1
class VcdSink: class VcdSink:
def __init__(self, fs, signals, timescale='1 us'): def __init__(self, fs, signals, timescale='1 us', verbose_trace=False):
self.writer = VCDWriter(fs, timescale=timescale, date=datetime.datetime.now().isoformat(), version=f"PET v1.0") self.writer = VCDWriter(fs, timescale=timescale, date=datetime.datetime.now().isoformat(), version=f"PET v1.0")
self.skalars = {} self.skalars = {}
self.arrays = {} self.arrays = {}
@ -348,6 +348,7 @@ class VcdSink:
self.varnames = {} self.varnames = {}
self.timestamp = 0 self.timestamp = 0
self.packets_dropped = 0 self.packets_dropped = 0
self.verbose_trace = verbose_trace
for v in signals: for v in signals:
hvar, vtype = v.split(":") hvar, vtype = v.split(":")
hier, _, name = hvar.rpartition(".") hier, _, name = hvar.rpartition(".")
@ -397,16 +398,17 @@ class VcdSink:
elif datatag[0] == 'A': elif datatag[0] == 'A':
timestamp = self.timestamp timestamp = self.timestamp
try: try:
#print(f"### {timestamp:012X} : {self.varnames[tag]}[{sub}] <= {value} [OK] ", flush=True) if self.verbose_trace:
print(f"### {timestamp:012} : {self.varnames[tag]}[{sub}] <= {value} [OK] ", flush=True)
self.writer.change(self.arrays[tag][sub], timestamp, value) self.writer.change(self.arrays[tag][sub], timestamp, value)
except ValueError: except ValueError:
print(f"### {timestamp:012X} : {self.varnames[tag]}[{sub}] <= {value} [VAL_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]}[{sub}] <= {value} [VAL_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except writer.VCDPhaseError: except writer.VCDPhaseError:
print(f"### {timestamp:012X} : {self.varnames[tag]}[{sub}] <= {value} [PHA_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]}[{sub}] <= {value} [PHA_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except: except:
print(f"### {timestamp:012X} : {self.varnames[tag]}[{sub}] <= {value} [ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]}[{sub}] <= {value} [ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
elif datatag == 'S4': elif datatag == 'S4':
timestamp = self.timestamp timestamp = self.timestamp
@ -419,16 +421,17 @@ class VcdSink:
if sub == 1: if sub == 1:
try: try:
string = self.strings[tag][1] string = self.strings[tag][1]
#print(f"### {timestamp:012X} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\"", flush=True) if self.verbose_trace:
self.writer.change(self.strings[tag][0], timestamp, self.strings[tag][1]) print(f"### {timestamp:012} : {self.varnames[tag]} <= \"{string}\"", flush=True)
self.writer.change(self.strings[tag][0], timestamp, string)
except ValueError: except ValueError:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [VAL_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [VAL_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except writer.VCDPhaseError: except writer.VCDPhaseError:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [PHA_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [PHA_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except: except:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= \"{self.strings[tag][1]}\" [ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
self.strings[tag][1] = "" self.strings[tag][1] = ""
@ -440,16 +443,17 @@ class VcdSink:
value = True value = True
elif datatag == 'F4': elif datatag == 'F4':
value = struct.unpack(">f", struct.pack(">L", value))[0] value = struct.unpack(">f", struct.pack(">L", value))[0]
#print(f"### {timestamp:012X} : {self.varnames[tag]} <= {value:08X}", flush=True) if self.verbose_trace:
print(f"### {timestamp:012} : {self.varnames[tag]} <= {value:08X}", flush=True)
self.writer.change(self.skalars[tag], timestamp, value) self.writer.change(self.skalars[tag], timestamp, value)
except ValueError: except ValueError:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= {value} [VAL_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= {value} [VAL_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except writer.VCDPhaseError: except writer.VCDPhaseError:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= {value} [PHA_ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= {value} [PHA_ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
except: except:
print(f"### {timestamp:012X} : {self.varnames[tag]} <= {value} [ERR] ", flush=True) print(f"### {timestamp:012} : {self.varnames[tag]} <= {value} [ERR] ", flush=True)
self.packets_dropped += 1 self.packets_dropped += 1
def process_noise(noisefile, b): def process_noise(noisefile, b):
@ -469,6 +473,8 @@ def main():
help='source tree to scan for trace marks') help='source tree to scan for trace marks')
parser.add_argument('--diagnostics', action=argparse.BooleanOptionalAction, parser.add_argument('--diagnostics', action=argparse.BooleanOptionalAction,
help='add additional signals tracing internal state of PET') help='add additional signals tracing internal state of PET')
parser.add_argument('--trace', action=argparse.BooleanOptionalAction,
help='write out every trace that arrives')
args = parser.parse_args() args = parser.parse_args()
print(header) print(header)
@ -478,6 +484,7 @@ def main():
source_tree = args.source source_tree = args.source
timescale = args.timescale timescale = args.timescale
enable_diag = args.diagnostics enable_diag = args.diagnostics
enable_verbose_trace = args.trace
predefined_signals = [] predefined_signals = []
if enable_diag: if enable_diag:
@ -506,7 +513,7 @@ def main():
nfile = open(noisefile, 'wb') nfile = open(noisefile, 'wb')
process_noise_p = partial(process_noise, nfile) process_noise_p = partial(process_noise, nfile)
vcd_sink = VcdSink(dfile, signals, timescale) vcd_sink = VcdSink(dfile, signals, timescale, enable_verbose_trace)
retagger = Retagger(vcd_sink.process, tags) retagger = Retagger(vcd_sink.process, tags)
packet_filter = Filter(retagger.process, process_noise_p) packet_filter = Filter(retagger.process, process_noise_p)
print("Signals:") print("Signals:")