added @timestamp decorator

Author: DinisCruz

osbot_utils/helpers/timestamp_capture/Timestamp_Collector.py

@@ -0,0 +1,99 @@
+"""
+Timestamp Collector - Core Recording
+=====================================
+
+Collects timestamps from @timestamp decorated methods.
+Found via stack-walking - no need to pass through call chain.
+
+Usage:
+    _timestamp_collector_ = Timestamp_Collector(name="my_workflow")
+    with _timestamp_collector_:
+        result = do_work()
+"""
+
+import time
+import threading
+from typing                                                                import List, Dict, Any
+from osbot_utils.helpers.timestamp_capture.schemas.Schema__Timestamp_Entry import Schema__Timestamp_Entry
+from osbot_utils.type_safe.Type_Safe                                       import Type_Safe
+
+
+class Timestamp_Collector(Type_Safe):
+
+    name            : str                              = 'default'
+    entries         : List[Schema__Timestamp_Entry]    = None
+    start_time_ns   : int                              = 0
+    end_time_ns     : int                              = 0
+    thread_id       : int                              = 0
+    _depth          : int                              = 0
+    _active         : bool                             = False
+    _call_stack     : List[Schema__Timestamp_Entry]    = None          # For self-time calculation
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.entries     = []
+        self._call_stack = []
+        self.thread_id   = threading.get_ident()
+
+    # ═══════════════════════════════════════════════════════════════════════════
+    # Context Manager
+    # ═══════════════════════════════════════════════════════════════════════════
+
+    def __enter__(self):
+        self.start_time_ns = time.perf_counter_ns()
+        self._active       = True
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.end_time_ns = time.perf_counter_ns()
+        self._active     = False
+        return False                                                            # Don't suppress exceptions
+
+    # ═══════════════════════════════════════════════════════════════════════════
+    # Recording API
+    # ═══════════════════════════════════════════════════════════════════════════
+
+    def record(self, name: str, event: str, extra: Dict[str, Any] = None) -> Schema__Timestamp_Entry:
+        if not self._active:                                                    # Record a timestamp entry (low-level API)
+            return None
+
+        entry = Schema__Timestamp_Entry(name         = name                   ,
+                                        event        = event                  ,
+                                        timestamp_ns = time.perf_counter_ns() ,
+                                        clock_ns     = time.time_ns()         ,
+                                        thread_id    = threading.get_ident()  ,
+                                        depth        = self._depth            ,
+                                        extra        = extra                  )
+        self.entries.append(entry)
+        return entry
+
+    def enter(self, name: str, extra: Dict[str, Any] = None):                   # Record method entry
+        entry = self.record(name, 'enter', extra)
+        if entry:
+            self._call_stack.append(entry)
+        self._depth += 1
+
+    def exit(self, name: str, extra: Dict[str, Any] = None):                    # Record method exit
+        self._depth = max(0, self._depth - 1)
+        self.record(name, 'exit', extra)
+        if self._call_stack:
+            self._call_stack.pop()
+
+    # ═══════════════════════════════════════════════════════════════════════════
+    # Basic Accessors
+    # ═══════════════════════════════════════════════════════════════════════════
+
+    def total_duration_ns(self) -> int:
+        return self.end_time_ns - self.start_time_ns
+
+    def total_duration_ms(self) -> float:
+        return self.total_duration_ns() / 1_000_000
+
+    def entry_count(self) -> int:
+        return len(self.entries)
+
+    def method_count(self) -> int:
+        return len(set(e.name for e in self.entries))
+
+    def is_active(self) -> bool:
+        return self._active

osbot_utils/helpers/timestamp_capture/Timestamp_Collector__Analysis.py

@@ -0,0 +1,67 @@
+"""
+Timestamp Collector Analysis
+=============================
+
+Analyzes collected timestamps to compute method timings and hotspots.
+"""
+
+from typing                                                                import List, Dict
+from osbot_utils.helpers.timestamp_capture.schemas.Schema__Method_Timing   import Schema__Method_Timing
+from osbot_utils.type_safe.Type_Safe                                       import Type_Safe
+from osbot_utils.helpers.timestamp_capture.Timestamp_Collector             import Timestamp_Collector
+
+
+class Timestamp_Collector__Analysis(Type_Safe):
+
+    collector: Timestamp_Collector = None
+
+    def get_method_timings(self) -> Dict[str, Schema__Method_Timing]:           # Calculate per-method timing with self-time
+        timings  : Dict[str, Schema__Method_Timing] = {}
+        stack    : List[tuple]                      = []                        # (entry_index, entry)
+        children : Dict[int, int]                   = {}                        # entry_index -> child_time_ns
+
+        for i, entry in enumerate(self.collector.entries):
+            if entry.event == 'enter':
+                stack.append((i, entry))
+                children[i] = 0
+
+            elif entry.event == 'exit' and stack:
+                for j in range(len(stack) - 1, -1, -1):                         # Find matching enter (reverse search)
+                    enter_idx, enter_entry = stack[j]
+                    if enter_entry.name == entry.name:
+                        stack.pop(j)
+
+                        duration_ns = entry.timestamp_ns - enter_entry.timestamp_ns
+                        child_time  = children.pop(enter_idx, 0)
+                        self_time   = duration_ns - child_time
+
+                        if stack:                                               # Add child time to parent
+                            parent_idx = stack[-1][0]
+                            children[parent_idx] = children.get(parent_idx, 0) + duration_ns
+
+                        if entry.name not in timings:                           # Update timing stats
+                            timings[entry.name] = Schema__Method_Timing(name   = entry.name  ,
+                                                                        min_ns = duration_ns ,
+                                                                        max_ns = duration_ns )
+
+                        mt             = timings[entry.name]
+                        mt.call_count += 1
+                        mt.total_ns   += duration_ns
+                        mt.self_ns    += self_time
+                        mt.min_ns      = min(mt.min_ns, duration_ns)
+                        mt.max_ns      = max(mt.max_ns, duration_ns)
+                        break
+
+        return timings
+
+    def get_hotspots(self, top_n: int = 10) -> List[Schema__Method_Timing]:     # Get top N methods by self-time
+        timings = self.get_method_timings()
+        return sorted(timings.values(), key=lambda t: t.self_ns, reverse=True)[:top_n]
+
+    def get_timings_by_total(self) -> List[Schema__Method_Timing]:              # Get all timings sorted by total time
+        timings = self.get_method_timings()
+        return sorted(timings.values(), key=lambda t: t.total_ns, reverse=True)
+
+    def get_timings_by_call_count(self) -> List[Schema__Method_Timing]:         # Get all timings sorted by call count
+        timings = self.get_method_timings()
+        return sorted(timings.values(), key=lambda t: t.call_count, reverse=True)

osbot_utils/helpers/timestamp_capture/Timestamp_Collector__Report.py

@@ -0,0 +1,119 @@
+"""
+Timestamp Collector Report
+===========================
+
+Formats and prints timing reports from collected timestamps.
+"""
+from osbot_utils.helpers.timestamp_capture.Timestamp_Collector import Timestamp_Collector
+from osbot_utils.helpers.timestamp_capture.Timestamp_Collector__Analysis         import Timestamp_Collector__Analysis
+from osbot_utils.helpers.timestamp_capture.static_methods.timestamp_utils        import method_timing__total_ms, method_timing__self_ms, method_timing__avg_ms
+from osbot_utils.type_safe.Type_Safe                                             import Type_Safe
+
+
+class Timestamp_Collector__Report(Type_Safe):
+
+    collector: Timestamp_Collector = None
+
+    def format_report(self, show_self_time: bool = True) -> str:                # Format comprehensive timing report
+        lines = []
+        lines.append("=" * 100)
+        lines.append(f"Timestamp Report: {self.collector.name}")
+        lines.append("=" * 100)
+        lines.append("")
+
+        total_ms = self.collector.total_duration_ms()
+        lines.append(f"  Total Duration : {total_ms:,.2f} ms")
+        lines.append(f"  Entry Count    : {self.collector.entry_count():,}")
+        lines.append(f"  Methods Traced : {self.collector.method_count()}")
+        lines.append("")
+
+        analysis = Timestamp_Collector__Analysis(collector=self.collector)
+        timings  = analysis.get_method_timings()
+
+        if timings:
+            lines.append("Method Timings (sorted by total time):")
+            lines.append("-" * 100)
+
+            if show_self_time:
+                header = f"{'Method':<50} {'Calls':>6} {'Total':>10} {'Self':>10} {'Avg':>8} {'%Total':>7}"
+            else:
+                header = f"{'Method':<50} {'Calls':>6} {'Total(ms)':>10} {'Avg(ms)':>10} {'%Total':>7}"
+
+            lines.append(header)
+            lines.append("-" * 100)
+
+            sorted_timings = analysis.get_timings_by_total()
+            total_ns       = self.collector.total_duration_ns()
+
+            for mt in sorted_timings:
+                pct = (mt.total_ns / total_ns * 100) if total_ns > 0 else 0
+
+                if show_self_time:
+                    lines.append(
+                        f"{mt.name:<50} {mt.call_count:>6} "
+                        f"{method_timing__total_ms(mt):>9.2f}ms {method_timing__self_ms(mt):>9.2f}ms "
+                        f"{method_timing__avg_ms(mt):>7.3f}ms {pct:>6.1f}%"
+                    )
+                else:
+                    lines.append(
+                        f"{mt.name:<50} {mt.call_count:>6} "
+                        f"{method_timing__total_ms(mt):>10.2f} {method_timing__avg_ms(mt):>10.3f} {pct:>6.1f}%"
+                    )
+
+        lines.append("")
+        lines.append("=" * 100)
+        return "\n".join(lines)
+
+    def format_timeline(self, max_entries: int = 100) -> str:                   # Format timeline view
+        lines = []
+        lines.append("=" * 80)
+        lines.append("Execution Timeline")
+        lines.append("=" * 80)
+
+        entries = self.collector.entries[:max_entries]
+        if len(self.collector.entries) > max_entries:
+            lines.append(f"(showing first {max_entries} of {len(self.collector.entries)} entries)")
+
+        for entry in entries:
+            offset_ms = (entry.timestamp_ns - self.collector.start_time_ns) / 1_000_000
+            indent    = "  " * entry.depth
+            marker    = "▶" if entry.event == 'enter' else "◀"
+            lines.append(f"{offset_ms:>10.3f}ms {indent}{marker} {entry.name}")
+
+        lines.append("=" * 80)
+        return "\n".join(lines)
+
+    def format_hotspots(self, top_n: int = 10) -> str:                          # Format hotspot analysis (by self-time)
+        lines = []
+        lines.append("=" * 80)
+        lines.append(f"Top {top_n} Hotspots (by self-time)")
+        lines.append("=" * 80)
+
+        analysis = Timestamp_Collector__Analysis(collector=self.collector)
+        hotspots = analysis.get_hotspots(top_n)
+        total_ns = self.collector.total_duration_ns()
+
+        for i, mt in enumerate(hotspots, 1):
+            pct = (mt.self_ns / total_ns * 100) if total_ns > 0 else 0
+            lines.append(
+                f"  {i:>2}. {mt.name:<45} "
+                f"{method_timing__self_ms(mt):>8.2f}ms ({pct:>5.1f}%) "
+                f"[{mt.call_count} calls]"
+            )
+
+        lines.append("=" * 80)
+        return "\n".join(lines)
+
+    def print_all(self):
+        self.print_report  ()
+        self.print_timeline()
+        self.print_hotspots()
+
+    def print_report(self, show_self_time: bool = True):
+        print(self.format_report(show_self_time))
+
+    def print_timeline(self, max_entries: int = 100):
+        print(self.format_timeline(max_entries))
+
+    def print_hotspots(self, top_n: int = 10):
+        print(self.format_hotspots(top_n))

osbot_utils/helpers/timestamp_capture/__init__.py

@@ -0,0 +1,29 @@
+"""
+Timestamp Capture - Type_Safe Integration
+==========================================
+
+Type_Safe compatible version of the timestamp capture system,
+designed for integration with the Html_MGraph pipeline.
+
+Key Features:
+- Inherits from Type_Safe for consistency with codebase
+- Stack-walking to find collector (no signature changes needed)
+- Minimal overhead when collector not present
+- Rich timing data with aggregation
+
+Integration Example:
+    # In Html__To__Html_MGraph__Document
+    @timestamp
+    def convert(self, html: str) -> Html_MGraph__Document:
+        ...
+
+    @timestamp
+    def _process_body(self, ...):
+        ...
+
+    # In test or profiling code
+    _timestamp_collector_ = Timestamp_Collector()
+    with _timestamp_collector_:
+        doc = converter.convert(html)
+    _timestamp_collector_.print_report()
+"""

osbot_utils/helpers/timestamp_capture/context_managers/__init__.py

@@ -0,0 +1,23 @@
+"""
+Context manager to timestamp a code block.
+
+Usage:
+    with timestamp_block("my_phase"):
+        # code to measure
+        ...
+"""
+from contextlib                                                                    import contextmanager
+from osbot_utils.helpers.timestamp_capture.static_methods.find_timestamp_collector import find_timestamp_collector
+
+
+@contextmanager
+def timestamp_block(name: str):
+
+    collector = find_timestamp_collector()
+    if collector:
+        collector.enter(name)
+    try:
+        yield
+    finally:
+        if collector:
+            collector.exit(name)

osbot_utils/helpers/timestamp_capture/context_managers/timestamp_block.py

@@ -0,0 +1,44 @@
+"""
+Timestamp Capture - @timestamp decorator
+==========================================
+
+  Usage:
+        @timestamp
+        def my_method(self):
+            ...
+
+        @timestamp(name="custom.name")
+        def another(self):
+            ...
+"""
+
+from functools                                                                     import wraps
+from typing                                                                        import Callable, Optional
+from osbot_utils.helpers.timestamp_capture.static_methods.find_timestamp_collector import find_timestamp_collector
+
+
+def timestamp(func: Callable = None, *, name: str = None):          #  Decorator to capture method timestamps.
+
+    def decorator(fn: Callable) -> Callable:
+        method_name = name or fn.__qualname__
+
+        @wraps(fn)
+        def wrapper(*args, **kwargs):
+            collector = find_timestamp_collector()
+
+            if collector is None:
+                return fn(*args, **kwargs)
+
+            collector.enter(method_name)
+            try:
+                return fn(*args, **kwargs)
+            finally:
+                collector.exit(method_name)
+
+        return wrapper
+
+    if func is not None:
+        return decorator(func)
+    return decorator
+
+