r2u2_cli 4.1.0

R2U2 CLI: A stream-based runtime monitor command-line interface
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

import pathlib
from typing import Optional
import re

from c2po import types, log

MODULE_CODE = "PRSO"

def parse_trace_file(
    trace_path: pathlib.Path, map_file_provided: bool
) -> tuple[int, Optional[types.SignalMapping]]:
    """Given `trace_path`, return the inferred length of the trace and, if `return_mapping` is enabled, a signal mapping."""
    with open(trace_path, "r") as f:
        content: str = f.read()

    lines: list[str] = content.splitlines()

    if len(lines) < 1:
        return (-1, None)

    cnt: int = 0
    signal_mapping: types.SignalMapping = {}

    if lines[0][0] == "#":
        # then there is a header
        header = lines[0][1:]

        if map_file_provided:
            log.warning(
                MODULE_CODE,
                "Map file given and header included in trace file; header will be ignored"
            )

        for id in [s.strip() for s in header.split(",")]:
            if id in signal_mapping:
                log.warning(
                    MODULE_CODE,
                    f"Signal ID '{id}' found multiple times in csv, using right-most value",
                    log.FileLocation(trace_path.name, 1)
                )
            signal_mapping[id] = cnt
            cnt += 1

        trace_length = len(lines) - 1

        return (trace_length, signal_mapping)

    # no header, so just return number of lines in file (i.e., number of time steps in trace)
    return (len(lines), None)


def parse_map_file(map_path: pathlib.Path) -> Optional[types.SignalMapping]:
    """Return the signal mapping from `map_path`."""
    with open(map_path, "r") as f:
        content: str = f.read()

    mapping: types.SignalMapping = {}

    lines = content.splitlines()
    for line in lines:
        if re.match(r"[a-zA-Z_][a-zA-Z0-9_\[\]]*:\d+", line):
            strs = line.split(":")
            id = strs[0]
            sid = int(strs[1])

            if id in mapping:
                log.warning(
                    MODULE_CODE,
                    f"Signal ID '{id}' found multiple times in map file, using latest value",
                    log.FileLocation(map_path.name, lines.index(line) + 1),
                )

            mapping[id] = sid
        else:
            log.error(
                MODULE_CODE,
                f"Invalid format for map line (found {line})"
                "\n\t Should be of the form SYMBOL ':' NUMERAL",
                log.FileLocation(map_path.name, lines.index(line)),
            )
            return None

    return mapping


def parse_bounds_c(bounds_path: pathlib.Path) -> None:
    pass


def parse_bounds_rust(bounds_path: pathlib.Path) -> None:
    pass