pytorch

1
import argparse
2
import concurrent.futures
3
import json
4
import logging
5
import os
6
import re
7
import subprocess
8
import time
9
from enum import Enum
10
from typing import List, NamedTuple, Optional, Pattern
11

12

13
LINTER_CODE = "CMAKE"
14

15

16
class LintSeverity(str, Enum):
17
    ERROR = "error"
18
    WARNING = "warning"
19
    ADVICE = "advice"
20
    DISABLED = "disabled"
21

22

23
class LintMessage(NamedTuple):
24
    path: Optional[str]
25
    line: Optional[int]
26
    char: Optional[int]
27
    code: str
28
    severity: LintSeverity
29
    name: str
30
    original: Optional[str]
31
    replacement: Optional[str]
32
    description: Optional[str]
33

34

35
# CMakeLists.txt:901: Lines should be <= 80 characters long [linelength]
36
RESULTS_RE: Pattern[str] = re.compile(
37
    r"""(?mx)
38
    ^
39
    (?P<file>.*?):
40
    (?P<line>\d+):
41
    \s(?P<message>.*)
42
    \s(?P<code>\[.*\])
43
    $
44
    """
45
)
46

47

48
def run_command(
49
    args: List[str],
50
) -> "subprocess.CompletedProcess[bytes]":
51
    logging.debug("$ %s", " ".join(args))
52
    start_time = time.monotonic()
53
    try:
54
        return subprocess.run(
55
            args,
56
            capture_output=True,
57
        )
58
    finally:
59
        end_time = time.monotonic()
60
        logging.debug("took %dms", (end_time - start_time) * 1000)
61

62

63
def check_file(
64
    filename: str,
65
    config: str,
66
) -> List[LintMessage]:
67
    try:
68
        proc = run_command(
69
            ["cmakelint", f"--config={config}", filename],
70
        )
71
    except OSError as err:
72
        return [
73
            LintMessage(
74
                path=None,
75
                line=None,
76
                char=None,
77
                code=LINTER_CODE,
78
                severity=LintSeverity.ERROR,
79
                name="command-failed",
80
                original=None,
81
                replacement=None,
82
                description=(f"Failed due to {err.__class__.__name__}:\n{err}"),
83
            )
84
        ]
85
    stdout = str(proc.stdout, "utf-8").strip()
86
    return [
87
        LintMessage(
88
            path=match["file"],
89
            name=match["code"],
90
            description=match["message"],
91
            line=int(match["line"]),
92
            char=None,
93
            code=LINTER_CODE,
94
            severity=LintSeverity.ERROR,
95
            original=None,
96
            replacement=None,
97
        )
98
        for match in RESULTS_RE.finditer(stdout)
99
    ]
100

101

102
if __name__ == "__main__":
103
    parser = argparse.ArgumentParser(
104
        description="cmakelint runner",
105
        fromfile_prefix_chars="@",
106
    )
107
    parser.add_argument(
108
        "--config",
109
        required=True,
110
        help="location of cmakelint config",
111
    )
112
    parser.add_argument(
113
        "filenames",
114
        nargs="+",
115
        help="paths to lint",
116
    )
117

118
    args = parser.parse_args()
119

120
    with concurrent.futures.ThreadPoolExecutor(
121
        max_workers=os.cpu_count(),
122
        thread_name_prefix="Thread",
123
    ) as executor:
124
        futures = {
125
            executor.submit(
126
                check_file,
127
                filename,
128
                args.config,
129
            ): filename
130
            for filename in args.filenames
131
        }
132
        for future in concurrent.futures.as_completed(futures):
133
            try:
134
                for lint_message in future.result():
135
                    print(json.dumps(lint_message._asdict()), flush=True)
136
            except Exception:
137
                logging.critical('Failed at "%s".', futures[future])
138
                raise
139