objwatch package

Subpackages

• objwatch.utils package
  • Submodules
  • objwatch.utils.logger module
    • create_logger()
    • get_logger()
    • log_debug()
    • log_error()
    • log_info()
    • log_warn()
  • objwatch.utils.weak module
    • WeakIdKeyDictionary
      • WeakIdKeyDictionary.copy()
      • WeakIdKeyDictionary.get()
      • WeakIdKeyDictionary.items()
      • WeakIdKeyDictionary.keyrefs()
      • WeakIdKeyDictionary.keys()
      • WeakIdKeyDictionary.pop()
      • WeakIdKeyDictionary.popitem()
      • WeakIdKeyDictionary.setdefault()
      • WeakIdKeyDictionary.update()
      • WeakIdKeyDictionary.values()
    • WeakIdRef
    • WeakTensorKeyDictionary
  • Module contents

Submodules

objwatch.core module

class objwatch.core.ObjWatch(targets: List[str | ModuleType], exclude_targets: List[str | ModuleType] | None = None, ranks: List[int] | None = None, output: str | None = None, output_xml: str | None = None, level: int = 10, simple: bool = False, wrapper: FunctionWrapper | None = None, with_locals: bool = False, with_globals: bool = False, with_module_path: bool = False)

Bases: object

Tracing and logging of specified Python modules to aid in debugging and monitoring.

load_wrapper(wrapper: FunctionWrapper) → FunctionWrapper

Load a custom wrapper into the tracer.

Args:

wrapper (FunctionWrapper): The custom wrapper to be loaded.

Returns:

FunctionWrapper: The result of the tracer’s load_wrapper method.

start() → None

Start the ObjWatch tracing process.

stop() → None

Stop the ObjWatch tracing process.

objwatch.core.watch(targets: List[str | ModuleType], exclude_targets: List[str | ModuleType] | None = None, ranks: List[int] | None = None, output: str | None = None, output_xml: str | None = None, level: int = 10, simple: bool = False, wrapper: FunctionWrapper | None = None, with_locals: bool = False, with_globals: bool = False, with_module_path: bool = False) → ObjWatch

Initialize and start an ObjWatch instance.

Args:

targets (List[Union[str, ModuleType]]): Files or modules to monitor. exclude_targets (Optional[List[Union[str, ModuleType]]]): Files or modules to exclude from monitoring. ranks (Optional[List[int]]): GPU ranks to track when using torch.distributed. output (Optional[str]): Path to a file for writing logs. output_xml (Optional[str]): Path to the XML file for writing structured logs. level (int): Logging level (e.g., logging.DEBUG, logging.INFO). simple (bool): Enable simple logging mode with the format “DEBUG: {msg}”. wrapper (Optional[FunctionWrapper]): Custom wrapper to extend tracing and logging functionality. with_locals (bool): Enable tracing and logging of local variables within functions. with_globals (bool): Enable tracing and logging of global variables across function calls. with_module_path (bool): Prepend the module path to function names in logs.

Returns:

ObjWatch: The initialized and started ObjWatch instance.

objwatch.event_handls module

class objwatch.event_handls.EventHandls(output_xml: str | None = None)

Bases: object

Handles various events for ObjWatch, including function execution and variable updates. Optionally saves the events in an XML format.

determine_change_type(old_value_len: int, current_value_len: int) → EventType

Determine the type of change based on the difference in lengths.

Args:

old_value_len (int): Previous length of the data structure. current_value_len (int): New length of the data structure.

Returns:

EventType: The determined event type (APD or POP).

static format_sequence(seq: Any, max_elements: int = 3, func: LambdaType | None = None) → str

Format a sequence to display a limited number of elements.

Args:

seq (Any): The sequence to format. max_elements (int): Maximum number of elements to display. func (Optional[FunctionType]): Optional function to process elements.

Returns:

str: The formatted sequence string.

handle_apd(lineno: int, class_name: str, key: str, value_type: type, old_value_len: int, current_value_len: int, call_depth: int, rank_info: str) → None

Handle the ‘apd’ event denoting the addition of elements to data structures.

Args:

lineno (int): The line number where the event is called. class_name (str): Name of the class containing the data structure. key (str): Name of the data structure. value_type (type): Type of the elements being added. old_value_len (int): Previous length of the data structure. current_value_len (int): New length of the data structure. call_depth (int): Current depth of the call stack. rank_info (str): Information about the GPU rank, if applicable.

handle_end(lineno: int, func_info: Dict[str, Any], function_wrapper: Any | None, call_depth: int, rank_info: str, result: Any) → None

Handle the ‘end’ event indicating the end of a function or method execution.

Args:

lineno (int): The line number where the event is called. func_info (Dict[str, Any]): Information about the function that has ended. function_wrapper (Optional[Any]): Custom wrapper for additional processing. call_depth (int): Current depth of the call stack. rank_info (str): Information about the GPU rank, if applicable. result (Any): The result returned by the function.

handle_pop(lineno: int, class_name: str, key: str, value_type: type, old_value_len: int, current_value_len: int, call_depth: int, rank_info: str) → None

Handle the ‘pop’ event marking the removal of elements from data structures.

Args:

lineno (int): The line number where the event is called. class_name (str): Name of the class containing the data structure. key (str): Name of the data structure. value_type (type): Type of the elements being removed. old_value_len (int): Previous length of the data structure. current_value_len (int): New length of the data structure. call_depth (int): Current depth of the call stack. rank_info (str): Information about the GPU rank, if applicable.

handle_run(lineno: int, func_info: Dict[str, Any], function_wrapper: Any | None, call_depth: int, rank_info: str) → None

Handle the ‘run’ event indicating the start of a function or method execution.

Args:

lineno (int): The line number where the event is called. func_info (Dict[str, Any]): Information about the function being executed. function_wrapper (Optional[Any]): Custom wrapper for additional processing. call_depth (int): Current depth of the call stack. rank_info (str): Information about the GPU rank, if applicable.

handle_upd(lineno: int, class_name: str, key: str, old_value: Any, current_value: Any, call_depth: int, rank_info: str, function_wrapper: Any | None = None) → None

Handle the ‘upd’ event representing the creation or updating of a variable.

Args:

lineno (int): The line number where the event is called. class_name (str): Name of the class containing the variable. key (str): Variable name. old_value (Any): Previous value of the variable. current_value (Any): New value of the variable. call_depth (int): Current depth of the call stack. rank_info (str): Information about the GPU rank, if applicable. function_wrapper (Optional[Any]): Custom wrapper for additional processing.

save_xml() → None

Save the accumulated events to an XML file upon program exit.

signal_handler(signum, frame)

Signal handler for abnormal program termination. Calls save_xml when a termination signal is received.

Args:

signum (int): The signal number. frame (frame): The current stack frame.

objwatch.events module

class objwatch.events.EventType(value)

Bases: Enum

Enumeration of event types used by ObjWatch to categorize tracing events.

APD = 4

END = 2

POP = 5

RUN = 1

UPD = 3

objwatch.tracer module

class objwatch.tracer.Tracer(targets: List[str | ModuleType], exclude_targets: List[str] | None = None, ranks: List[int] | None = None, wrapper: FunctionWrapper | None = None, output_xml: str | None = None, with_locals: bool = False, with_globals: bool = False, with_module_path: bool = False)

Bases: object

Tracer class to monitor and trace function calls, returns, and variable updates within specified target modules. Supports multi-GPU environments with PyTorch.

load_wrapper(wrapper: FunctionWrapper | None) → FunctionWrapper | None

Load a custom function wrapper if provided.

Args:

wrapper (Optional[FunctionWrapper]): The custom wrapper to load.

Returns:

Optional[FunctionWrapper]: The initialized wrapper or None.

start() → None

Start the tracing process by setting the trace function.

stop() → None

Stop the tracing process by removing the trace function and saving XML logs.

trace_factory() → LambdaType

Create the tracing function to be used with sys.settrace.

Returns:

FunctionType: The trace function.

objwatch.wrappers module

class objwatch.wrappers.BaseLogger

Bases: FunctionWrapper

BaseLogger implements the FunctionWrapper abstract methods to provide basic logging functionality.

wrap_call(func_name: str, frame: FrameType) → str

Format the function call information.

Args:

func_name (str): Name of the function being called. frame (FrameType): The current stack frame.

Returns:

str: Formatted call message.

wrap_return(func_name: str, result: Any) → str

Format the function return information.

Args:

func_name (str): Name of the function returning. result (Any): The result returned by the function.

Returns:

str: Formatted return message.

wrap_upd(old_value: Any, current_value: Any) → Tuple[str, str]

Format the update information of a variable.

Args:

old_value (Any): The old value of the variable. current_value (Any): The new value of the variable.

Returns:

Tuple[str, str]: Formatted old and new values.

class objwatch.wrappers.FunctionWrapper

Bases: ABC

Abstract base class for function wrappers to extend tracing and logging functionality.

abstract wrap_call(func_name: str, frame: FrameType) → str

Process and format the function call information.

Args:

func_name (str): Name of the function being called. frame (FrameType): The current stack frame.

Returns:

str: Formatted call message.

abstract wrap_return(func_name: str, result: Any) → str

Process and format the function return information.

Args:

func_name (str): Name of the function returning. result (Any): The result returned by the function.

Returns:

str: Formatted return message.

abstract wrap_upd(old_value: Any, current_value: Any) → Tuple[str, str]

Process and format the update information of a variable.

Args:

old_value (Any): The old value of the variable. current_value (Any): The new value of the variable.

Returns:

Tuple[str, str]: Formatted old and new values.

class objwatch.wrappers.TensorShapeLogger

Bases: FunctionWrapper

TensorShapeLogger extends FunctionWrapper to log the shapes of torch.Tensor objects.

wrap_call(func_name: str, frame: FrameType) → str

Format the function call information, including tensor shapes if applicable.

Args:

func_name (str): Name of the function being called. frame (FrameType): The current stack frame.

Returns:

str: Formatted call message.

wrap_return(func_name: str, result: Any) → str

Format the function return information, including tensor shapes if applicable.

Args:

func_name (str): Name of the function returning. result (Any): The result returned by the function.

Returns:

str: Formatted return message.

wrap_upd(old_value: Any, current_value: Any) → Tuple[str, str]

Format the update information of a variable, including tensor shapes if applicable.

Args:

old_value (Any): The old value of the variable. current_value (Any): The new value of the variable.

Returns:

Tuple[str, str]: Formatted old and new values.

Module contents

ObjWatch Package

A Python library to trace and monitor object attributes and method calls.

Exports:

ObjWatch: The main class for setting up and managing tracing. watch: A convenience function to start tracing with default settings.

class objwatch.ObjWatch(targets: List[str | ModuleType], exclude_targets: List[str | ModuleType] | None = None, ranks: List[int] | None = None, output: str | None = None, output_xml: str | None = None, level: int = 10, simple: bool = False, wrapper: FunctionWrapper | None = None, with_locals: bool = False, with_globals: bool = False, with_module_path: bool = False)

Bases: object

Tracing and logging of specified Python modules to aid in debugging and monitoring.

load_wrapper(wrapper: FunctionWrapper) → FunctionWrapper

Load a custom wrapper into the tracer.

Args:

wrapper (FunctionWrapper): The custom wrapper to be loaded.

Returns:

FunctionWrapper: The result of the tracer’s load_wrapper method.

start() → None

Start the ObjWatch tracing process.

stop() → None

Stop the ObjWatch tracing process.

objwatch.watch(targets: List[str | ModuleType], exclude_targets: List[str | ModuleType] | None = None, ranks: List[int] | None = None, output: str | None = None, output_xml: str | None = None, level: int = 10, simple: bool = False, wrapper: FunctionWrapper | None = None, with_locals: bool = False, with_globals: bool = False, with_module_path: bool = False) → ObjWatch

Initialize and start an ObjWatch instance.

Args:

targets (List[Union[str, ModuleType]]): Files or modules to monitor. exclude_targets (Optional[List[Union[str, ModuleType]]]): Files or modules to exclude from monitoring. ranks (Optional[List[int]]): GPU ranks to track when using torch.distributed. output (Optional[str]): Path to a file for writing logs. output_xml (Optional[str]): Path to the XML file for writing structured logs. level (int): Logging level (e.g., logging.DEBUG, logging.INFO). simple (bool): Enable simple logging mode with the format “DEBUG: {msg}”. wrapper (Optional[FunctionWrapper]): Custom wrapper to extend tracing and logging functionality. with_locals (bool): Enable tracing and logging of local variables within functions. with_globals (bool): Enable tracing and logging of global variables across function calls. with_module_path (bool): Prepend the module path to function names in logs.

Returns:

ObjWatch: The initialized and started ObjWatch instance.