Source code for napari.utils.perf._event

"""PerfEvent class."""

import os
import threading
from typing import NamedTuple, Optional


class Span(NamedTuple):
    """The span of time that the event ocurred.

    Parameters
    ----------
    start_ns : int
        Start time in nanoseconds.
    end_ns : int
        End time in nanoseconds.
    """

    start_ns: int
    end_ns: int


class Origin(NamedTuple):
    """What process/thread produced the event.

    Parameters
    ----------
    process_id : int
        The process id that produced the event.
    thread_id : int
        The thread id that produced the event.
    """

    process_id: int
    thread_id: int


[docs] class PerfEvent: """A performance related event: timer, counter, etc. Parameters ---------- name : str The name of this event like "draw". start_ns : int Start time in nanoseconds. end_ns : int End time in nanoseconds. category :str Comma separated categories such has "render,update". process_id : int The process id that produced the event. thread_id : int The thread id that produced the event. phase : str The Chrome Tracing "phase" such as "X", "I", "C". **kwargs : dict Additional keyword arguments for the "args" field of the event. Attributes ---------- name : str The name of this event like "draw". span : Span The time span when the event happened. category : str Comma separated categories such as "render,update". origin : Origin The process and thread that produced the event. args : dict Arbitrary keyword arguments for this event. phase : str The Chrome Tracing phase (event type): "X" - Complete Events "I" - Instant Events "C" - Counter Events Notes ----- The time stamps are from perf_counter_ns() and do not indicate time of day. The origin is arbitrary, but subtracting two counters results in a valid span of wall clock time. If start is the same as the end the event was instant. Google the phrase "Trace Event Format" for the full Chrome Tracing spec. """ def __init__( self, name: str, start_ns: int, end_ns: int, category: Optional[str] = None, process_id: Optional[int] = None, thread_id: Optional[int] = None, phase: str = 'X', # "X" is a "complete event" in their spec. **kwargs: float, ) -> None: if process_id is None: process_id = os.getpid() if thread_id is None: thread_id = threading.get_ident() self.name: str = name self.span: Span = Span(start_ns, end_ns) self.category: Optional[str] = category self.origin: Origin = Origin(process_id, thread_id) self.args = kwargs self.phase: str = phase
[docs] def update_end_ns(self, end_ns: int) -> None: """Update our end_ns with this new end_ns. Attributes ---------- end_ns : int The new ending time in nanoseconds. """ self.span = Span(self.span.start_ns, end_ns)
@property def start_us(self) -> float: """Start time in microseconds.""" return self.span.start_ns / 1e3 @property def start_ms(self) -> float: """Start time in milliseconds.""" return self.span.start_ns / 1e6 @property def duration_ns(self) -> int: """Duration in nanoseconds.""" return self.span.end_ns - self.span.start_ns @property def duration_us(self) -> float: """Duration in microseconds.""" return self.duration_ns / 1e3 @property def duration_ms(self) -> float: """Duration in milliseconds.""" return self.duration_ns / 1e6