mirrors/sby
3
0
Fork 0
mirror of https://github.com/YosysHQ/sby.git synced 2025-08-20 20:00:25 +00:00
Code Activity

statusfmt: Make JSONL self-contained and escape CSV values

Browse source
This commit is contained in:
Jannis Harder 2025-07-29 17:10:57 +02:00
parent f05979a528
commit 344236af41
1 changed files with 42 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

51
sbysrc/sby_status.py
View file

@ -448,6 +448,7 @@ class SbyStatusDb:
# print header # print header
header = format_status_data_fmtline(None, status_format) header = format_status_data_fmtline(None, status_format)
if header:
print(header) print(header)
# find summary for each task/property combo # find summary for each task/property combo
@ -506,9 +507,7 @@ def parse_status_data_row(raw: sqlite3.Row):
row_dict["data"] = json.loads(row_dict.get("data") or "{}") row_dict["data"] = json.loads(row_dict.get("data") or "{}")
return row_dict return row_dict
def format_status_data_fmtline(row: dict|None, fmt: str = "csv") -> str: fmtline_columns = [
if row is None:
data = [
"time", "time",
"task_name", "task_name",
"mode", "mode",
@ -519,7 +518,11 @@ def format_status_data_fmtline(row: dict|None, fmt: str = "csv") -> str:
"status", "status",
"trace", "trace",
"depth", "depth",
] ]
def format_status_data_fmtline(row: dict|None, fmt: str = "csv") -> str:
if row is None:
data = None
else: else:
engine = row['data'].get('engine', row['data'].get('source')) engine = row['data'].get('engine', row['data'].get('source'))
try: try:
@ -533,22 +536,34 @@ def format_status_data_fmtline(row: dict|None, fmt: str = "csv") -> str:
except TypeError: except TypeError:
trace_path = None trace_path = None
csv_line = [ data = {
round(time, 2), "time": round(time, 2),
row['task_name'], "task_name": row['task_name'],
row['mode'], "mode": row['mode'],
engine, "engine": engine,
name or pretty_path(row['name']), "name": name or pretty_path(row['name']),
row['location'], "location": row['location'],
row['kind'], "kind": row['kind'],
row['status'] or "UNKNOWN", "status": row['status'] or "UNKNOWN",
trace_path, "trace": trace_path,
depth, "depth": depth,
] }
data = ["" if v is None else str(v) for v in csv_line]
if fmt == "csv": if fmt == "csv":
return ','.join(data) if data is None:
csv_line = fmtline_columns
else:
csv_line = [data[column] for column in fmtline_columns]
def csv_field(value):
if value is None:
return ""
value = str(value).replace('"', '""')
if any(c in value for c in '",\n'):
value = f'"{value}"'
return value
return ','.join(map(csv_field, csv_line))
elif fmt == "jsonl": elif fmt == "jsonl":
if data is None:
return ""
return json.dumps(data) return json.dumps(data)
def filter_latest_task_ids(all_tasks: dict[int, dict[str]]): def filter_latest_task_ids(all_tasks: dict[int, dict[str]]):