Merge pull request #207 from willmoffat/main

new: kbdebug.txt parser

Author: cvandeplas

pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
   "jinja2>=3.1.0,<3.2.0",
   "matplotlib>=3.10.0,<3.11.0",
   "orjsonl>=1.0.0,<1.1.0",
+  "pyparsing>=3.2, <3.3",
 ]
 
 

src/sysdiagnose/parsers/kbdebug.py

@@ -0,0 +1,52 @@
+#! /usr/bin/env python3
+
+import os
+from sysdiagnose.utils.base import BaseParserInterface, SysdiagnoseConfig, logger, Event
+from pyparsing import Regex, SkipTo, StringEnd
+
+from datetime import datetime
+
+
+class KbdebugParser(BaseParserInterface):
+    description = "kbdebug.txt logfile parser"
+    format = "jsonl"  # by default json, use jsonl for event-based data
+
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
+        super().__init__(__file__, config, case_id)
+
+    def get_log_files(self) -> list:
+        log_files = [
+            "kbdebug.txt"
+        ]
+        return [os.path.join(self.case_data_subfolder, log_files) for log_files in log_files]
+
+    def execute(self) -> list | dict:
+        fmt_timestamp = Regex(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \+\d{4}")
+        fmt_level = Regex(r"\w+")
+        fmt_line = fmt_timestamp("timestamp") + ": " + fmt_level("level") + ": " + SkipTo(fmt_timestamp | StringEnd(), include=False)("message")
+
+        result = []
+        log_files = self.get_log_files()
+        for log_file in log_files:
+            with open(log_file, 'r') as f:
+                lines = f.read()
+                for tokens, _start, _end in fmt_line.scanString(lines):
+                    # print(f"Timestamp: {tokens.timestamp}")
+                    # print(f"Level: {tokens.level}")
+                    # print(f"Message: {tokens.message}")
+                    try:
+                        timestamp = datetime.strptime(tokens.timestamp, '%Y-%m-%d %H:%M:%S %z')
+                        event = Event(
+                            datetime=timestamp,
+                            message=tokens.message,  # The rest of the kbdebug entry
+                            module=self.module_name,
+                            timestamp_desc='kbdebug event',
+                            data={
+                                "level": tokens.level
+                            }
+                        )
+                        result.append(event.to_dict())
+                    except Exception:
+                        logger.exception("Failed to parse timestamp or create event")
+
+        return result

tests/test_parsers_kbdebug.py

@@ -0,0 +1,25 @@
+from sysdiagnose.parsers.kbdebug import KbdebugParser
+from tests import SysdiagnoseTestCase
+import unittest
+import os
+
+
+class TestParsersSys(SysdiagnoseTestCase):
+
+    def test_parsekbdebug(self):
+        for case_id, _case in self.sd.cases().items():
+            p = KbdebugParser(self.sd.config, case_id=case_id)
+            files = p.get_log_files()
+            self.assertTrue(len(files) > 0)
+
+            p.save_result(force=True)
+            self.assertTrue(os.path.isfile(p.output_file))
+
+            result = p.get_result()
+            self.assertGreater(len(result), 0)
+            for item in result:
+                self.assert_has_required_fields_jsonl(item)
+
+
+if __name__ == '__main__':
+    unittest.main()