Coverage for src/signalk_cli/history/output.py: 98%

1"""CSV, JSON and Feather writers for the history CLI.""" 

2 

3import csv 

4import json 

5from typing import IO, Any 

6 

7from .._arrow import as_text 

8from ._results import CARDINALITY_COLUMNS, long_rows, wide_rows 

9 

10 

11def _text(v: Any) -> str: 

12 """CLI text for a value: JSON for objects/arrays, empty for null.""" 

13 return as_text(v) or "" 

14 

15 

16# --------------------------------------------------------------------------- 

17# Long (timestamp, path, value) 

18# --------------------------------------------------------------------------- 

19 

20 

21def write_csv(result: dict, sink: IO[str], no_header: bool) -> tuple[int, set[str]]: 

22 """Write result as CSV rows (timestamp, path, value). Returns (row_count, unique_paths).""" 

23 timestamps, paths, values = long_rows(result) 

24 writer = csv.writer(sink) 

25 if not no_header: 

26 writer.writerow(["timestamp", "path", "value"]) 

27 for ts, path, val in zip(timestamps, paths, values): 

28 writer.writerow([ts, path, _text(val)]) 

29 return len(timestamps), set(paths) 

30 

31 

32def write_json( 

33 result: dict, sink: IO[str], indent: int | None = None 

34) -> tuple[int, set[str]]: 

35 """Write result as a JSON array of {timestamp, path, value} objects.""" 

36 timestamps, paths, values = long_rows(result) 

37 rows = [ 

38 {"timestamp": ts, "path": p, "value": _text(v)} 

39 for ts, p, v in zip(timestamps, paths, values) 

40 ] 

41 sink.write(json.dumps(rows, indent=indent)) 

42 return len(rows), set(paths) 

43 

44 

45# --------------------------------------------------------------------------- 

46# Wide (timestamp, path, min_value/avg_value/max_value or array element columns) 

47# --------------------------------------------------------------------------- 

48 

49 

50def write_csv_wide( 

51 result: dict, sink: IO[str], no_header: bool 

52) -> tuple[int, set[str]]: 

53 """Write result as CSV with wide value columns.""" 

54 timestamps, paths, value_cols = wide_rows(result) 

55 col_names = list(value_cols) 

56 writer = csv.writer(sink) 

57 if not no_header: 57 ↛ 59line 57 didn't jump to line 59 because the condition on line 57 was always true

58 writer.writerow(["timestamp", "path", *col_names]) 

59 for i, (ts, path) in enumerate(zip(timestamps, paths)): 

60 writer.writerow([ts, path, *(_text(value_cols[c][i]) for c in col_names)]) 

61 return len(timestamps), set(paths) 

62 

63 

64def write_json_wide( 

65 result: dict, sink: IO[str], indent: int | None = None 

66) -> tuple[int, set[str]]: 

67 """Write result as a JSON array of wide row objects.""" 

68 timestamps, paths, value_cols = wide_rows(result) 

69 rows = [ 

70 {"timestamp": ts, "path": p, **{c: _text(v[i]) for c, v in value_cols.items()}} 

71 for i, (ts, p) in enumerate(zip(timestamps, paths)) 

72 ] 

73 sink.write(json.dumps(rows, indent=indent)) 

74 return len(rows), set(paths) 

75 

76 

77# --------------------------------------------------------------------------- 

78# Cardinality 

79# --------------------------------------------------------------------------- 

80 

81 

82def cardinality_text(row: dict[str, Any]) -> dict[str, str]: 

83 """Cardinality row as CLI text, with empty strings for stats that don't apply.""" 

84 return {col: _text(row[col]) for col in CARDINALITY_COLUMNS}