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.

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 elapse since start until end.

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.

Source code in pygx/instrument/_timing.py

def end(self, error: BaseException | None = None) -> bool:
    """Ends timing."""
    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 under this context.

Source code in pygx/instrument/_timing.py

def status(self) -> dict[str, Status]:
    """Gets the status of all `timeit` under this context."""
    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.

Source code in pygx/instrument/_timing.py

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

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