pygx.instrument

PyGX's instrumentation sub-packages.

instrument

PyGX's instrumentation sub-packages.

Three private modules back the public namespaces re-exported here:

• _logging is exposed as pg.logging / pygx.instrument.logging — use those names to set/read the project logger.
• _monitoring is exposed as pg.monitoring / pygx.instrument.monitoring for metric collection.
• _timing is flattened into this package: pg.instrument.TimeIt and pg.instrument.timeit are the public surface.

TimeIt

TimeIt(name: str = '')

Context manager for timing the execution of a code block.

TimeIt contexts nest per-thread: a context entered inside another becomes its child, and status() reports elapsed times for the whole subtree with dot-joined names. Usually created via pg.timeit.

Source code in pygx/instrument/_timing.py

def __init__(self, name: str = ''):
    self._name: str = name
    self._start_time: float | None = None
    self._end_time: float | None = None
    self._child_contexts: list[TimeIt] = []
    self._error: error_info.ErrorInfo | None = None
    self._parent: TimeIt | None = None

name property

name: str

Returns the name of the context.

children property

children: list[TimeIt]

Returns child contexts.

has_started property

has_started: bool

Returns whether the context has started.

has_ended property

has_ended: bool

Returns whether the context has ended.

start_time property

start_time: float | None

Returns start time.

end_time property

end_time: float | None

Returns end time.

error property

error: ErrorInfo | None

Returns error.

has_error property

has_error: bool

Returns whether the context has error.

elapse property

elapse: float

Returns the elapsed seconds from start to end.

Returns 0 when the context has not started; when it has started but not ended, returns the elapsed time up to now.

Status dataclass

Status(
    name: str,
    elapse: float = 0.0,
    has_ended: bool = True,
    error: ErrorInfo | None = None,
)

Bases: WireConvertible

Status of a single pg.timeit.

has_started property

has_started: bool

Returns whether the context has started.

has_error property

has_error: bool

Returns whether the context has error.

merge

merge(other: Status) -> Status

Merges the status of two pg.timeit.

Source code in pygx/instrument/_timing.py

def merge(self, other: 'TimeIt.Status') -> 'TimeIt.Status':
    """Merges the status of two `pg.timeit`."""
    assert other.name == self.name, (self.name, other.name)
    return TimeIt.Status(
        name=self.name,
        elapse=self.elapse + other.elapse,
        has_ended=self.has_ended and other.has_ended,
        error=self.error or other.error,
    )

StatusSummary dataclass

StatusSummary(breakdown: dict[str, Entry] = dict())

Bases: WireConvertible

Aggregated summary for repeated calls for pg.timeit.

Entry dataclass

Entry(
    num_started: int = 0,
    num_ended: int = 0,
    num_failed: int = 0,
    avg_duration: float = 0.0,
    error_tags: dict[str, int] = (lambda: defaultdict(int))(),
)

Bases: WireConvertible

Aggregated status from the pg.timeit calls of the same name.

add

add(context: TimeIt)

Adds a child context.

Source code in pygx/instrument/_timing.py

def add(self, context: 'TimeIt'):
    """Adds a child context."""
    self._child_contexts.append(context)

start

start()

Starts timing.

Source code in pygx/instrument/_timing.py

def start(self):
    """Starts timing."""
    self._start_time = time.time()

end

end(error: BaseException | None = None) -> bool

Ends timing; returns False if the context had already ended.

Source code in pygx/instrument/_timing.py

def end(self, error: BaseException | None = None) -> bool:
    """Ends timing; returns False if the context had already ended."""
    if not self.has_ended:
        self._end_time = time.time()
        self._error = (
            None
            if error is None
            else error_info.ErrorInfo.from_exception(error)
        )
        return True
    return False

status

status() -> dict[str, Status]

Gets the status of all timeit contexts under this context.

Keys are dot-joined context names (e.g. 'job.load' for a child named 'load' under this context named 'job'); statuses of same-named children are merged.

Source code in pygx/instrument/_timing.py

def status(self) -> dict[str, Status]:
    """Gets the status of all ``timeit`` contexts under this context.

    Keys are dot-joined context names (e.g. ``'job.load'`` for a child
    named ``'load'`` under this context named ``'job'``); statuses of
    same-named children are merged.
    """
    result = {
        self.name: TimeIt.Status(
            name=self.name,
            elapse=self.elapse,
            has_ended=self.has_ended,
            error=self._error,
        )
    }
    for child in self._child_contexts:
        child_result = child.status()
        for k, v in child_result.items():
            key = f'{self.name}.{k}' if self.name else k
            if key in result:
                result[key] = result[key].merge(v)
            else:
                result[key] = v
    return result

timeit

timeit(name: str = '') -> TimeIt

Context manager to time a block of code.

Contexts nest per-thread: entering pg.timeit inside another makes it a child, and status() on the outer context reports elapsed times for the whole subtree keyed by dot-joined names.

Example:

with pg.timeit('job') as t:
  with pg.timeit('load'):
    ...
  with pg.timeit('compute'):
    ...

status = t.status()
# status keys: 'job', 'job.load', 'job.compute'.
print({name: s.elapse for name, s in status.items()})

Parameters:

Name	Type	Description	Default
name	str	Name of the timing context, used as (part of) its key in status().	''

Returns:

Type	Description
TimeIt	A TimeIt context manager object.

Source code in pygx/instrument/_timing.py

def timeit(name: str = '') -> TimeIt:
    """Context manager to time a block of code.

    Contexts nest per-thread: entering ``pg.timeit`` inside another makes it
    a child, and ``status()`` on the outer context reports elapsed times for
    the whole subtree keyed by dot-joined names.

    **Example:**

    ```python
    with pg.timeit('job') as t:
      with pg.timeit('load'):
        ...
      with pg.timeit('compute'):
        ...

    status = t.status()
    # status keys: 'job', 'job.load', 'job.compute'.
    print({name: s.elapse for name, s in status.items()})
    ```

    Args:
      name: Name of the timing context, used as (part of) its key in
        ``status()``.

    Returns:
      A [`TimeIt`][pygx.instrument.TimeIt] context manager object.
    """
    return TimeIt(name)

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2