updates, switch python to pypy for ~6x speedup

ise/ise_logparser/.python-version

@@ -1 +1 @@
-3.12.3
+pypy@3.10.14

ise/ise_logparser/pyproject.toml

@@ -9,6 +9,10 @@ dependencies = [
     "parsy>=2.1",
     "icecream>=2.1.3",
     "python-dotenv>=1.0.1",
+    "mypy>=1.11.1",
+    "setuptools>=72.1.0",
+    "colorama>=0.4.6",
+    "types-colorama>=0.4.15.20240311",
 ]
 readme = "README.md"
 requires-python = ">= 3.8"

ise/ise_logparser/requirements-dev.lock

@@ -14,15 +14,28 @@ asttokens==2.4.1
     # via icecream
 colorama==0.4.6
     # via icecream
+    # via ise-logparser
 executing==2.0.1
     # via icecream
 icecream==2.1.3
     # via ise-logparser
+mypy==1.11.1
+    # via ise-logparser
+mypy-extensions==1.0.0
+    # via mypy
 parsy==2.1
     # via ise-logparser
 pygments==2.18.0
     # via icecream
 python-dotenv==1.0.1
     # via ise-logparser
+setuptools==72.1.0
+    # via ise-logparser
 six==1.16.0
     # via asttokens
+tomli==2.0.1
+    # via mypy
+types-colorama==0.4.15.20240311
+    # via ise-logparser
+typing-extensions==4.12.2
+    # via mypy

ise/ise_logparser/requirements.lock

@@ -14,15 +14,28 @@ asttokens==2.4.1
     # via icecream
 colorama==0.4.6
     # via icecream
+    # via ise-logparser
 executing==2.0.1
     # via icecream
 icecream==2.1.3
     # via ise-logparser
+mypy==1.11.1
+    # via ise-logparser
+mypy-extensions==1.0.0
+    # via mypy
 parsy==2.1
     # via ise-logparser
 pygments==2.18.0
     # via icecream
 python-dotenv==1.0.1
     # via ise-logparser
+setuptools==72.1.0
+    # via ise-logparser
 six==1.16.0
     # via asttokens
+tomli==2.0.1
+    # via mypy
+types-colorama==0.4.15.20240311
+    # via ise-logparser
+typing-extensions==4.12.2
+    # via mypy

ise/ise_logparser/src/main.py

@@ -1,93 +1,225 @@
 #!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import logging
+import os
 import re
 from pathlib import Path
-from dotenv import dotenv_values
+from typing import cast
+
 import parsy as ps
-from icecream import ic
+from colorama import Back, Fore, Style
-import os
+from dotenv import dotenv_values
-from pprint import pprint as print
+from parsy import Parser, regex, seq, string
-from parsy import regex, seq
 
-# TODO: (akristip) refactor these to use `<<` instead of capture groups
+LEVEL_UNKNOWN = 255
-parse_header = regex(
+NEWLINE = string("\n")
-    r"(.*)(^[^\n]*-+\n.*EASYPY JOB START.*?-+\|\n)",
+SPACE = string(" ")
-    re.MULTILINE | re.DOTALL,
-    1,
-)
-parse_test_output = regex(
-    r"(.*)(^[^\n]*-+\n.*EASYPY JOB END.*?-+\|)",
-    re.MULTILINE | re.DOTALL,
-    1,
-)
-parse_post_test = regex(
-    r"(.*)(\+-+\+\n\| +Easypy Report +\|\n\+-+\+)",
-    re.MULTILINE | re.DOTALL,
-    1,
-)
-parse_report = regex(
-    r"(.*)(\+-+\+\n\| +Task Result Summary +\|\n\+-+\+)",
-    re.MULTILINE | re.DOTALL,
-    1,
-)
-parse_summary = regex(
-    r"(.*)(\+-+\+\n\| +Task Result Details +\|\n\+-+\+)",
-    re.MULTILINE | re.DOTALL,
-    1,
-)
-
-timestamp = regex(r"\d{4}-\d{2}-\d{2}.\d{2}:\d{2}:\d{2}(,\d{3})?")
-log_level = regex(r" (.+?) ", 0, 1)
-source = regex(r"\[([A-Za-z_]+)\]", 0, 1)
-messsage = ps.any_char.until(ps.peek(timestamp) | ps.eof)
-
-
-parse_details = regex(r".*", re.MULTILINE | re.DOTALL)
-parse_sections = seq(
-    header=parse_header,
-    test_output=parse_test_output,
-    post_test=parse_post_test,
-    report=parse_report,
-    summary=parse_summary,
-    details=parse_details,
-)
-parse_lines = seq(
-    ts=timestamp,
-    level=log_level,
-    src=source,
-    msg=messsage.map(lambda x: "".join(x).strip()),
-).many()
 
 
 class Config(dict[str, str]):
-    def __init__(self):
+    class LevelAction(argparse.Action):
-        self.dict = dotenv_values(".env")
+        def __call__(self, parser, namespace, values, option_string=None) -> None:
+            _ = (parser, option_string)
+            values = cast(str, values)
+            setattr(namespace, self.dest, format_debug_level(values.upper())[0])
+
+    def __init__(self) -> None:
+        self.dict = {k: v for k, v in dotenv_values(".env").items() if v is not None}
+        parser = argparse.ArgumentParser()
+        if not ("LOGFILE" in self.dict or "LOGFILE" in os.environ):
+            parser.add_argument(
+                "LOGFILE",
+                type=Path,
+            )
+        parser.add_argument("-q", "--query", dest="QUERY")
+        parser.add_argument(
+            "-l",
+            "--level",
+            dest="LOGLEVEL",
+            type=str,
+            choices=["debug", "info", "warn", "err", "crit", "unk"],
+            default=logging.INFO,
+            action=self.LevelAction,
+        )
+
+        self.dict.update(parser.parse_args().__dict__)
 
     def __getitem__(self, key: str, /) -> str:
         try:
-            return self.dict.__getitem__(key)
+            return self.dict[key]
         except KeyError:
             return os.environ[key]
 
-    def __setitem__(self, _key: str, _value: str, /):
+    def __setitem__(self, _key: str, _value: str, /) -> None:
-        raise NotImplementedError("Config is read-only")
+        msg = "Config is read-only"
+        raise NotImplementedError(msg)
 
-    def __delitem__(self, _: str, /):
+    def __delitem__(self, _: str, /) -> None:
-        raise NotImplementedError("Config is read-only")
+        msg = "Config is read-only"
+        raise NotImplementedError(msg)
 
     def __iter__(self):
         return iter(self.dict)
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self.dict)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return self.dict.__repr__()
 
 
+"""
+Formatting functions
+"""
+
+
+def colorize_status(status: str) -> str:
+    if "PASS" in status:
+        return Fore.GREEN + status + Style.RESET_ALL
+    elif "FAIL" in status:
+        return Fore.RED + status + Style.RESET_ALL
+    elif "ERROR" in status:
+        return Fore.BLACK + Back.RED + status + Style.RESET_ALL
+    return status
+
+
+def format_debug_level(level: str) -> tuple[int, str]:
+    if level == "DEBUG":
+        return (logging.DEBUG, level)
+    elif level == "INFO":
+        return (logging.INFO, level)
+    elif "WARN" in level:
+        return (logging.WARNING, level)
+    elif "ERR" in level:
+        return (logging.ERROR, level)
+    elif "CRIT" in level:
+        return (logging.CRITICAL, level)
+    else:
+        return (LEVEL_UNKNOWN, level)
+
+
+def colorize_by_level(level: str, text: str) -> str:
+    if level == "DEBUG":
+        return Style.DIM + text + Style.RESET_ALL
+    elif level == "INFO":
+        return Style.BRIGHT + text + Style.RESET_ALL
+    elif level == "WARNING":
+        return Fore.YELLOW + text + Style.RESET_ALL
+    elif level == "ERROR":
+        return Fore.RED + text + Style.RESET_ALL
+    elif level == "CRITICAL":
+        return Style.BRIGHT + Back.RED + Fore.BLACK + text + Style.RESET_ALL
+    else:
+        return Fore.BLUE + text + Style.RESET_ALL
+    # return level
+
+
+"""
+Parser defs
+"""
+
+
+def block_parser(text: str) -> Parser:
+    return regex(rf"^.*\n^.*{text}.*$\n.*", re.MULTILINE)
+
+
+timestamp: Parser = regex(r"\d{4}-\d{2}-\d{2}.\d{2}:\d{2}:\d{2}(,\d{3})?")
+debug_level: Parser = regex(r"\S+")
+
+
+start_block = block_parser("EASYPY JOB START")
+end_block = block_parser("EASYPY JOB END")
+report_block = block_parser("Easypy Report")
+summary_block = block_parser("Task Result Summary")
+details_block = block_parser("Task Result Details")
+
+
+sections = seq(
+    header=ps.any_char.until(start_block).concat() << start_block << NEWLINE,
+    test_output=ps.any_char.until(end_block).concat() << end_block << NEWLINE,
+    post_test=ps.any_char.until(report_block).concat() << report_block,
+    report=ps.any_char.until(summary_block).concat() << summary_block,
+    summary=ps.any_char.until(details_block).concat() << details_block,
+    details=ps.any_char.until(NEWLINE + ps.letter).concat() << NEWLINE,
+    etc=ps.any_char.until(ps.eof).concat(),
+)
+messsage = ps.any_char.until(ps.peek(timestamp) | ps.eof).concat()
+test_output = seq(
+    ts=timestamp << SPACE,
+    level=debug_level.map(format_debug_level) << SPACE,
+    src=string("[") >> regex(r"[A-Za-z_0-9]+") << string("] "),
+    msg=messsage.map(lambda x: x.strip()),
+).many()
+
+
+test_details: Parser = seq(
+    test_name=(
+        NEWLINE.optional()
+        << (SPACE * 4)
+        << (string("|--") | string("`--"))
+        << ps.whitespace
+    )
+    >> ps.any_char.until(ps.whitespace).concat()
+    << ps.whitespace,
+    test_results=(ps.letter.many().concat()).map(colorize_status),
+)
+
+subtask_details: Parser = seq(
+    subtask_name=(string("`--") << ps.whitespace)
+    >> ps.any_char.until(ps.whitespace).concat()
+    << ps.whitespace,
+    subtask_result=(ps.letter.many().concat()).map(colorize_status),
+    tests=test_details.many(),
+)
+
+task_details: Parser = seq(
+    task_number=ps.any_char.until(string(":")).concat() << string(":") << ps.whitespace,
+    task_name=ps.any_char.until(ps.whitespace).concat() << ps.whitespace,
+    subtasks=subtask_details.many(),
+)
+details: Parser = task_details.many()
+
+
+"""
+Output functions
+"""
+
+
+def output_details(log_details: list) -> None:
+    for task in log_details:
+        print(
+            f"{task['task_number']}\t {Style.BRIGHT + task['task_name'] + Style.RESET_ALL}"
+        )
+        for subtask in task["subtasks"]:
+            print(f"  {subtask['subtask_result']}\t{subtask['subtask_name']}")
+            for test in subtask["tests"]:
+                print(f" |  {test['test_results']}\t{test['test_name']}")
+
+
+def output_tests(
+    tests: list,
+    filter_level: int = logging.NOTSET,
+    search_term: str | re.Pattern | None = None,
+) -> None:
+    for test in tests:
+        level = test["level"]
+        if (search_term and not re.search(search_term, test["msg"])) or level[
+            0
+        ] >= filter_level:
+            print(
+                colorize_by_level(
+                    level[1], f"{test['ts']} {level[1]} {test['src']} {test['msg']}"
+                )
+            )
+
+
 if __name__ == "__main__":
     config = Config()
     with Path(config["LOGFILE"]).open() as f:
         log = f.read()
-    sections = parse_sections.parse(log)
+    parsed_sections = sections.parse(log)
-    lines = parse_lines.parse(sections["test_output"])
+    output_details(details.parse(parsed_sections["details"]))
-    ic(lines)
+    print("-------------\v")
+    output_tests(test_output.parse(parsed_sections["test_output"]), config["LOGLEVEL"])  # noqa