Minimap WIP

src/toolong/format_parser.py

@@ -40,13 +40,16 @@ def parse(self, line: str) -> ParseResult | None:
         _, timestamp = timestamps.parse(groups["date"].strip("[]"))
 
         text = self.highlighter(line)
-        if status := groups.get("status", None):
-            text.highlight_words(
-                [f" {status} "], "bold red" if status.startswith("4") else "magenta"
-            )
+        if status := groups.get("status", "").strip():
+            if status.startswith("4"):
+                text.highlight_words([f" {status} "], "bold #ffa62b")
+            elif status.startswith("5"):
+                text.highlight_words([f" {status} "], "bold white on red")
+            else:
+                text.highlight_words([f" {status} "], "bold magenta")
         text.highlight_words(self.HIGHLIGHT_WORDS, "bold yellow")
 
-        return timestamp, line, text
+        return timestamp, line, text, (status.startswith("4"))
 
 
 class CommonLogFormat(RegexLogFormat):
@@ -90,7 +93,7 @@ def parse(self, line: str) -> ParseResult | None:
     JSONLogFormat(),
     CommonLogFormat(),
     CombinedLogFormat(),
-    DefaultLogFormat(),
+    # DefaultLogFormat(),
 ]
 
 
@@ -109,4 +112,4 @@ def parse(self, line: str) -> ParseResult:
                     del self._formats[index : index + 1]
                     self._formats.insert(0, format)
                 return parse_result
-        return None, "", Text()
+        return None, "", Text(), False

src/toolong/log_lines.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass
 from queue import Empty, Queue
-from threading import Event, Thread
+from threading import Event, Lock, Thread
 
 from textual.message import Message
 from textual.suggester import Suggester
@@ -46,7 +46,7 @@
 
 
 @dataclass
-class LineRead(Message):
+class LineRead(Message, verbose=True):
     """A line has been read from the file."""
 
     index: int
@@ -86,7 +86,7 @@ def run(self) -> None:
         log_lines = self.log_lines
         while not self.exit_event.is_set():
             try:
-                request = self.queue.get(timeout=0.2)
+                request = self.queue.get(timeout=1)
             except Empty:
                 continue
             else:
@@ -96,13 +96,7 @@ def run(self) -> None:
                 if self.exit_event.is_set() or log_file is None:
                     break
                 log_lines.post_message(
-                    LineRead(
-                        index,
-                        log_file,
-                        start,
-                        end,
-                        log_file.get_line(start, end),
-                    )
+                    LineRead(index, log_file, start, end, log_file.get_line(start, end))
                 )
 
 
@@ -150,17 +144,17 @@ class LogLines(ScrollView, inherit_bindings=False):
     LogLines {
         scrollbar-gutter: stable;
         overflow: scroll;
-        border: heavy transparent;        
+        # border: heavy transparent;        
         .loglines--filter-highlight {
             background: $secondary;
             color: auto;
         }
         .loglines--pointer-highlight {
             background: $primary;
         }
-        &:focus {
-            border: heavy $accent;
-        }
+        # &:focus {
+        #     border: heavy $accent;
+        # }
 
         border-subtitle-color: $success;
         border-subtitle-align: center;        
@@ -221,6 +215,8 @@ def __init__(self, watcher: Watcher, file_paths: list[str]) -> None:
         self._gutter_width = 0
         self._line_reader = LineReader(self)
         self._merge_lines: list[tuple[float, int, LogFile]] | None = None
+        self._errors: list[int] = []
+        self._lock = Lock()
 
     @property
     def log_file(self) -> LogFile:
@@ -248,6 +244,17 @@ def clear_caches(self) -> None:
         self._line_cache.clear()
         self._text_cache.clear()
 
+    def add_error(self, offset: int) -> None:
+        """Add error line
+
+        Args:
+            offset: Offset within the file.
+        """
+        error_offset = offset // 8
+        if error_offset > len(self._errors):
+            self._errors.extend([0] * 100)
+        self._errors[error_offset] += 1
+
     def notify_style_update(self) -> None:
         self.clear_caches()
 
@@ -409,20 +416,21 @@ def get_log_file_from_index(self, index: int) -> tuple[LogFile, int]:
         return self.log_files[0], index
 
     def index_to_span(self, index: int) -> tuple[LogFile, int, int]:
-        log_file, index = self.get_log_file_from_index(index)
-        line_breaks = self._line_breaks.setdefault(log_file, [])
-        if not line_breaks:
-            return (log_file, self._scan_start, self._scan_start)
-        index = clamp(index, 0, len(line_breaks))
-        if index == 0:
-            return (log_file, self._scan_start, line_breaks[0])
-        start = line_breaks[index - 1]
-        end = (
-            line_breaks[index]
-            if index < len(line_breaks)
-            else max(0, self._scanned_size - 1)
-        )
-        return (log_file, start, end)
+        with self._lock:
+            log_file, index = self.get_log_file_from_index(index)
+            line_breaks = self._line_breaks.setdefault(log_file, [])
+            if not line_breaks:
+                return (log_file, self._scan_start, self._scan_start)
+            index = clamp(index, 0, len(line_breaks))
+            if index == 0:
+                return (log_file, self._scan_start, line_breaks[0])
+            start = line_breaks[index - 1]
+            end = (
+                line_breaks[index]
+                if index < len(line_breaks)
+                else max(0, self._scanned_size - 1)
+            )
+            return (log_file, start, end)
 
     def get_line_from_index_blocking(self, index: int) -> str:
         log_file, start, end = self.index_to_span(index)
@@ -476,7 +484,7 @@ def get_text(
             if new_line is None:
                 return "", Text(""), None
             line = new_line
-            timestamp, line, text = log_file.parse(line)
+            timestamp, line, text, error = log_file.parse(line)
             if abbreviate and len(text) > MAX_LINE_LENGTH:
                 text = text[:MAX_LINE_LENGTH] + "…"
             self._text_cache[cache_key] = (line, text, timestamp)

src/toolong/log_view.py

@@ -29,6 +29,7 @@
 )
 from toolong.find_dialog import FindDialog
 from toolong.line_panel import LinePanel
+from toolong.mini_map import Minimap
 from toolong.watcher import Watcher
 from toolong.log_lines import LogLines
 
@@ -252,6 +253,16 @@ class LogView(Horizontal):
             width: 50%;
             display: none;            
         }
+        #log-container {
+            border: heavy transparent;
+            &:focus-within {
+                border: heavy $accent;
+            }
+            Minimap {
+                margin-left: 1;
+                padding: 0 0 1 0;
+            }
+        }
     }
     """
 
@@ -280,14 +291,16 @@ def __init__(
         self.call_later(setattr, self, "can_tail", can_tail)
 
     def compose(self) -> ComposeResult:
-        yield (
-            log_lines := LogLines(self.watcher, self.file_paths).data_bind(
+        with Horizontal(id="log-container"):
+            log_lines = LogLines(self.watcher, self.file_paths).data_bind(
                 LogView.tail,
                 LogView.show_line_numbers,
                 LogView.show_find,
                 LogView.can_tail,
             )
-        )
+            yield log_lines
+            yield Minimap(log_lines)
+
         yield LinePanel()
         yield FindDialog(log_lines._suggester)
         yield InfoOverlay().data_bind(LogView.tail)
@@ -398,6 +411,7 @@ async def on_scan_complete(self, event: ScanComplete) -> None:
 
         footer = self.query_one(LogFooter)
         footer.call_after_refresh(footer.mount_keys)
+        self.query_one(Minimap).refresh_map(log_lines._line_count)
 
     @on(events.DescendantFocus)
     @on(events.DescendantBlur)

src/toolong/map_renderable.py

@@ -0,0 +1,67 @@
+from rich.console import Console, ConsoleOptions, RenderResult
+from rich.segment import Segment
+from rich.style import Style
+
+from textual.color import Color, Gradient
+
+
+COLORS = [
+    "#881177",
+    "#aa3355",
+    "#cc6666",
+    "#ee9944",
+    "#eedd00",
+    "#99dd55",
+    "#44dd88",
+    "#22ccbb",
+    "#00bbcc",
+    "#0099cc",
+    "#3366bb",
+    "#663399",
+]
+
+gradient = Gradient(
+    (0.0, Color.parse("transparent")),
+    (0.01, Color.parse("#004578")),
+    (0.8, Color.parse("#FF7043")),
+    (1.0, Color.parse("#ffaa43")),
+)
+
+
+class MapRenderable:
+
+    def __init__(self, data: list[int], height: int) -> None:
+        self._data = data
+        self._height = height
+
+    def __rich_console__(
+        self, console: Console, options: ConsoleOptions
+    ) -> RenderResult:
+        width = options.max_width
+        height = self._height
+
+        step = (len(self._data) / height) / 2
+        data = [
+            sum(self._data[round(step_no * step) : round(step_no * step + step)])
+            for step_no in range(height * 2)
+        ]
+
+        max_value = max(data)
+        get_color = gradient.get_color
+        style_from_color = Style.from_color
+
+        for datum1, datum2 in zip(data[::2], data[1::2]):
+            value1 = (datum1 / max_value) if max_value else 0
+            color1 = get_color(value1).rich_color
+            value2 = (datum2 / max_value) if max_value else 0
+            color2 = get_color(value2).rich_color
+            yield Segment(f"{'▀' * width}\n", style_from_color(color1, color2))
+
+
+if __name__ == "__main__":
+
+    from rich import print
+
+    map = MapRenderable([1, 4, 0, 0, 10, 4, 3, 6, 1, 0, 0, 0, 12, 10, 11, 0], 2)
+
+    print(map)

src/toolong/messages.py

@@ -84,3 +84,8 @@ class PointerMoved(Message):
 
     def can_replace(self, message: Message) -> bool:
         return isinstance(message, PointerMoved)
+
+
+@dataclass
+class MinimapUpdate(Message):
+    data: list[int]

src/toolong/mini_map.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import rich.repr
+
+from textual.app import ComposeResult
+from textual import work, on
+from textual.worker import get_current_worker
+from textual.message import Message
+from textual.reactive import reactive
+from textual.widget import Widget
+from textual.widgets import Static
+
+from toolong.map_renderable import MapRenderable
+
+
+if TYPE_CHECKING:
+    from toolong.log_lines import LogLines
+
+
+class Minimap(Widget):
+
+    DEFAULT_CSS = """
+    Minimap {
+        width: 3;
+        height: 1fr;
+    }
+
+    """
+
+    data: reactive[list[int]] = reactive(list, always_update=True)
+
+    @dataclass
+    @rich.repr.auto
+    class UpdateData(Message):
+        data: list[int]
+
+        def __rich_repr__(self) -> rich.repr.Result:
+            yield self.data[:10]
+
+    def __init__(self, log_lines: LogLines) -> None:
+        self._log_lines = log_lines
+        super().__init__()
+
+    @on(UpdateData)
+    def update_data(self, event: UpdateData) -> None:
+        self.data = event.data
+
+    def render(self) -> MapRenderable:
+        return MapRenderable(self.data or [0, 0], self.size.height)
+
+    def refresh_map(self, line_count: int) -> None:
+        self.scan_lines(self.data.copy(), 0, line_count)
+
+    @work(thread=True, exclusive=True)
+    def scan_lines(self, data: list[int], start_line: int, end_line: int) -> None:
+        worker = get_current_worker()
+        line_no = start_line
+
+        data = [0] * (((end_line - start_line) + 7) // 8)
+        while line_no < end_line and not worker.is_cancelled:
+
+            log_file, start, end = self._log_lines.index_to_span(line_no)
+            line = log_file.get_line(start, end)
+            *_, error = log_file.format_parser.parse(line)
+
+            if error:
+                data[line_no // 8] += 1
+
+            line_no += 1
+
+        if worker.is_cancelled:
+            return
+        self.post_message(self.UpdateData(data))