nd2py.utils package

class nd2py.utils.AttrDict(*args, **kwargs)

Bases: dict

__init__(*args, **kwargs)

classmethod load_yaml(_AttrDict__filepath)

classmethod load_yaml_str(str)

class nd2py.utils.EqualizeNormalize(*args: Any, **kwargs: Any)

Bases: Normalize

按分布而非值进行归一化

__init__(samples, clip=False)

inverse(value)

nd2py.utils.MAE_score(true, pred)

nd2py.utils.MAPE_score(true, pred)

class nd2py.utils.NamedTimer(unit='iter')

Bases: Timer

__init__(unit='iter')

对 Timer 的扩展, 支持将 count 和 time 统计到不同的名称下

add(name, n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace', mode_of_detail: Literal['time', 'count', 'pace', 'speed'] | None = 'pace', mode_of_percent: Literal['by_time', 'by_count'] | None = 'by_time')

property names

property time

property count

get_named_time(name)

get_named_count(name)

get_named_pace(name)

get_named_speed(name)

property named_time

property named_count

property named_pace

property named_speed

class nd2py.utils.ParallelTimer(unit='iter')

Bases: Timer

__init__(unit='iter')

对 Timer 的扩展, 支持将 count 统计到不同的名称下, 但 time 在各名称间共享

add(name, n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace', mode_of_detail: Literal['time', 'count', 'pace', 'speed'] | None = 'pace', mode_of_percent: Literal['by_time', 'by_count'] | None = 'by_time')

property names

property count

get_named_time(name)

get_named_count(name)

get_named_pace(name)

get_named_speed(name)

property named_time

property named_count

property named_pace

property named_speed

nd2py.utils.R2_score(true, pred)

nd2py.utils.RMSE_score(true, pred)

class nd2py.utils.Timer(unit='iter')

Bases: object

__init__(unit='iter')

计时器

add(n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace')

property time

property count

property pace

property speed

time_str()

count_str()

pace_str()

speed_str()

to_dict()

classmethod from_dict(dict)

nd2py.utils.add_minus_flags(parser: ArgumentParser)

自动为 argparse.ArgumentParser 中含有下划线的参数添加别名。例如： –fix_existing -> 添加别名 –fix-existing –augment_OD_num -> 添加别名 –augment-OD-num

nd2py.utils.add_negation_flags(parser: ArgumentParser)

自动为 parser 中的 store_true 参数添加对应的 –no-xxx 选项。

nd2py.utils.adjust_text(texts, ax, step=0.01, max_iterations=100, mode='xy')

class nd2py.utils.classproperty(fget)

Bases: object

自定义的类属性装饰器。 允许通过 ClassName.property_name 直接获取动态计算的类属性， 同时也支持 instance.property_name。

__init__(fget)

nd2py.utils.clear_svg(path, debug=False)

matplotlib 生成的 svg 中会使用 <text style=”font: 9.8px ‘Arial’; text-anchor: middle” x=”80.307802” y=”193.900483”>2020-02-02</text> 的语法，而 Powerpoint 无法识别 font: 9.8px ‘Arial’; 的简写记法，只能识别 font-family: ‘Arial’; font-size: 9.8px; 的记法。因此需要进行转换。考虑的属性包括： - font-size - font-family - font-weight - font-style

nd2py.utils.get_fig(RN, CN, FW=None, FH=None, AW=None, AH=None, A_ratio=1.0, LM=3, RM=3, TM=3, BM=3, HS=None, VS=None, dpi=300, fontsize=7, lw=0.5, gridspec=False, **kwargs)

nd2py.utils.init_logger(package_name: str, exp_name: str = None, log_file: str = None, info_level: Literal['debug', 'trace', 'info', 'note', 'warning', 'error', 'critical'] = 'info', file_max_size_MB: float = 50.0, file_backup_count: int = 100, show_lineno_for_all_levels: bool = False)

Initialize the logger for the package. Args: - package_name: The name of the package / project, used in loggin.getLogger({package_name}.{path}.{to}.{file}). - exp_name: The name of the experiment, used in log prefix. - log_file: The path to the log file. If None, no file logging is performed. - info_level: The level of info logging. Can be one of ‘debug’, ‘info’, ‘note’, ‘warning’, ‘error’, ‘critical’. - file_max_size_MB: The maximum size of the log file in MB. Default is 50MB. - file_backup_count: The number of backup files to keep. Default is 100.

nd2py.utils.load_font()

• plt.title(“示例图表：数字平方”, fontproperties=font, size=15)

• plt.xlabel(“数字”, fontproperties=font, size=12)

• plt.ylabel(“平方”, fontproperties=font, size=12)

nd2py.utils.log_exception(e: Exception) → str

Format exception for logging.

nd2py.utils.merge_axes(axes, fig=None)

nd2py.utils.plotOD(ax, source: List[str], destination: List[str], flow: List[float], location: Dict[str, Tuple[float, float]], linetype: Literal['straight', 'parabola', 'rotated_parabola', 'projected_parabola'] = 'straight', N=100, zorder=10, **kwargs)

绘制OD流量

nd2py.utils.plot_resilience(f: callable, extent=(0, 1, 0, 1), gridnum=(1000, 1000), cmap=None, norm=None, ax=None, reset_xylim=True, lw=1)

绘制二维函数的韧性 - f: 二维函数，如 lambda x, y: y - 3*y**2 - y**3 + x*y**3 - extent: 函数定义域 (xmin, xmax, ymin, ymax) - gridnum: 网格数量 (xnum, ynum) — Example: >>> f = lambda x, y: y - 3*y**2 - y**3 + x*y**3 >>> plot_resilience(f, ax=axes[0], extent=(0, 5, 0, 5)) >>> f = lambda x, y: np.sin(np.sqrt(x**2 + y ** 2)) >>> plot_resilience(f, ax=axes[1], extent=(-4*np.pi, 4*np.pi, -4*np.pi, 4*np.pi)) >>> f = lambda x, y: np.sin(np.sqrt(x ** 2 + y ** 2)) * x - y * np.cos(np.sqrt(x ** 2 + y ** 2)) >>> plot_resilience(f, ax=axes[2], extent=(-4*np.pi, 4*np.pi, -4*np.pi, 4*np.pi))

nd2py.utils.render_markdown(text: str, width=120, theme='staroffice') → str

nd2py.utils.render_python(text: str, width=120, highlight_lines=[], theme='staroffice') → str

nd2py.utils.sMAPE_score(true, pred)

nd2py.utils.seed_all(seed)

nd2py.utils.setup_lazy_imports(module_name: str, import_mapping: Dict[str, Tuple[str, str]])

Set up lazy imports for a module’s __init__.py.

Returns (__getattr__, __dir__, __all__) which should be assigned at the module level so that from package import OptionalClass works without importing the optional dependency until it is actually needed.

Parameters:

• module_name – The __name__ of the calling module.

• import_mapping – A dict mapping attribute names to (module_path, requires) tuples. module_path is a relative import path (e.g. ".torch_calc") and requires is the optional-dependency group name (e.g. "nn") shown in the error message when the dependency is missing.

Usage:

# __init__.py
from .core import CoreClass
from ..utils.lazy_loader import setup_lazy_imports

if TYPE_CHECKING:
    from .optional import OptionalClass

__getattr__, __dir__, __all__ = setup_lazy_imports(__name__, {
    "OptionalClass": (".optional", "nn"),
})

nd2py.utils.softmax(x)

nd2py.utils.tag2ansi(text: str) → str

将 [tag1 tag2] 转换为 ANSI 码并智能插入 reset。 支持多种颜色定义方式、样式组合、自动闭合。

nd2py.utils.use_chinese_font(fontpath='/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc')

Subpackages

• nd2py.utils.nn package
  • setup_lazy_imports()
  • Submodules
  • nd2py.utils.nn.gnn module
    • GNNLayer
      • GNNLayer.__init__()
      • GNNLayer.forward()
    • GNN
      • GNN.__init__()
      • GNN.fwd()
      • GNN.forward()
  • nd2py.utils.nn.positional_encoding module
    • PositionalEncoding
      • PositionalEncoding.__init__()
      • PositionalEncoding.forward()

Submodules

nd2py.utils.attr_dict module

class nd2py.utils.attr_dict.AttrDict(*args, **kwargs)

Bases: dict

__init__(*args, **kwargs)

classmethod load_yaml(_AttrDict__filepath)

classmethod load_yaml_str(str)

nd2py.utils.auto_gpu module

class nd2py.utils.auto_gpu.AutoGPU

Bases: object

Automatically choose a GPU with enough free memory

__init__()

update_free_memory()

choice_gpu(memory_MB, interval=600, force=True)

Choose a GPU with enough free memory Arguments: - memory_MB: int, the required memory in MB - interval: int, the interval (in second) to check free memory - force: bool, whether to wait until a GPU is available

query_free_memory(gpu_id)

nd2py.utils.classproperty module

class nd2py.utils.classproperty.classproperty(fget)

Bases: object

自定义的类属性装饰器。 允许通过 ClassName.property_name 直接获取动态计算的类属性， 同时也支持 instance.property_name。

__init__(fget)

nd2py.utils.factory module

class nd2py.utils.factory.FactoryMixin

Bases: object

Mixin class that provides factory pattern capabilities for any class.

Any class inheriting from this mixin gains the ability to: 1. Register subclasses via @<Class>.register_model(‘name’) 2. Create instances via <Class>.create(config, *args, **kwargs)

The base class is automatically registered as ‘default’ model.

═══════════════════════════════════════════════════════════════════════════ INHERITANCE ORDER (IMPORTANT) ═══════════════════════════════════════════════════════════════════════════

FactoryMixin should be placed BEFORE other base classes in the inheritance:

class MyModel(FactoryMixin, nn.Module):

…

This ensures correct Method Resolution Order (MRO) so that: 1. create() method resolves correctly 2. Subclasses share the same MODEL_DICT through the mixin

═══════════════════════════════════════════════════════════════════════════ EXAMPLE ═══════════════════════════════════════════════════════════════════════════

```python from nd2py.utils import FactoryMixin

class MyModel(FactoryMixin, nn.Module):

def __init__(self, config, tokenizer):

super().__init__() self.config = config

@MyModel.register_model(‘gcn’) class GCNModel(MyModel):

def __init__(self, config, tokenizer):

super().__init__(config, tokenizer) # … custom architecture

# Usage - create() automatically handles model selection config = MyConfig(model=’gcn’) model = MyModel.create(config, tokenizer)

# ‘default’ is automatically the base class config = MyConfig(model=’default’) model = MyModel.create(config, tokenizer) # Returns MyModel instance ```

MODEL_DICT: Dict[str, Type] = {'flash_ansr': <class 'nd2py.search.ndformer.ndformer_model_flash_ansr.FlashANSRNDFormer'>}

Registry of model classes keyed by name. Shared across all subclasses.

classmethod register_model(name: str)

Decorator to register a model subclass.

Usage:

@MyModel.register_model(‘gcn’) class GCNModel(MyModel):

…

classmethod create(config, *args, **kwargs) → T

Factory method to create instance based on config.model.

Parameters:

• config – Configuration object with model type specified in config.model

• *args – Positional arguments passed to constructor

• **kwargs – Keyword arguments passed to constructor

Returns:

Instantiated object of the type specified in config.model

Raises:

ValueError – If config.model is not found in MODEL_DICT

nd2py.utils.fix_parser module

nd2py.utils.fix_parser.add_minus_flags(parser: ArgumentParser)

自动为 argparse.ArgumentParser 中含有下划线的参数添加别名。例如： –fix_existing -> 添加别名 –fix-existing –augment_OD_num -> 添加别名 –augment-OD-num

nd2py.utils.fix_parser.add_negation_flags(parser: ArgumentParser)

自动为 parser 中的 store_true 参数添加对应的 –no-xxx 选项。

nd2py.utils.lazy_loader module

nd2py.utils.lazy_loader.setup_lazy_imports(module_name: str, import_mapping: Dict[str, Tuple[str, str]])

Set up lazy imports for a module’s __init__.py.

Returns (__getattr__, __dir__, __all__) which should be assigned at the module level so that from package import OptionalClass works without importing the optional dependency until it is actually needed.

Parameters:

• module_name – The __name__ of the calling module.

• import_mapping – A dict mapping attribute names to (module_path, requires) tuples. module_path is a relative import path (e.g. ".torch_calc") and requires is the optional-dependency group name (e.g. "nn") shown in the error message when the dependency is missing.

Usage:

# __init__.py
from .core import CoreClass
from ..utils.lazy_loader import setup_lazy_imports

if TYPE_CHECKING:
    from .optional import OptionalClass

__getattr__, __dir__, __all__ = setup_lazy_imports(__name__, {
    "OptionalClass": (".optional", "nn"),
})

nd2py.utils.log_exception module

nd2py.utils.log_exception.log_exception(e: Exception) → str

Format exception for logging.

nd2py.utils.logger module

nd2py.utils.logger.init_logger(package_name: str, exp_name: str = None, log_file: str = None, info_level: Literal['debug', 'trace', 'info', 'note', 'warning', 'error', 'critical'] = 'info', file_max_size_MB: float = 50.0, file_backup_count: int = 100, show_lineno_for_all_levels: bool = False)

Initialize the logger for the package. Args: - package_name: The name of the package / project, used in loggin.getLogger({package_name}.{path}.{to}.{file}). - exp_name: The name of the experiment, used in log prefix. - log_file: The path to the log file. If None, no file logging is performed. - info_level: The level of info logging. Can be one of ‘debug’, ‘info’, ‘note’, ‘warning’, ‘error’, ‘critical’. - file_max_size_MB: The maximum size of the log file in MB. Default is 50MB. - file_backup_count: The number of backup files to keep. Default is 100.

nd2py.utils.metrics module

nd2py.utils.metrics.R2_score(true, pred)

nd2py.utils.metrics.RMSE_score(true, pred)

nd2py.utils.metrics.sMAPE_score(true, pred)

nd2py.utils.metrics.MAPE_score(true, pred)

nd2py.utils.metrics.MAE_score(true, pred)

nd2py.utils.plot module

nd2py.utils.plot.get_fig(RN, CN, FW=None, FH=None, AW=None, AH=None, A_ratio=1.0, LM=3, RM=3, TM=3, BM=3, HS=None, VS=None, dpi=300, fontsize=7, lw=0.5, gridspec=False, **kwargs)

nd2py.utils.plot.plot_resilience(f: callable, extent=(0, 1, 0, 1), gridnum=(1000, 1000), cmap=None, norm=None, ax=None, reset_xylim=True, lw=1)

绘制二维函数的韧性 - f: 二维函数，如 lambda x, y: y - 3*y**2 - y**3 + x*y**3 - extent: 函数定义域 (xmin, xmax, ymin, ymax) - gridnum: 网格数量 (xnum, ynum) — Example: >>> f = lambda x, y: y - 3*y**2 - y**3 + x*y**3 >>> plot_resilience(f, ax=axes[0], extent=(0, 5, 0, 5)) >>> f = lambda x, y: np.sin(np.sqrt(x**2 + y ** 2)) >>> plot_resilience(f, ax=axes[1], extent=(-4*np.pi, 4*np.pi, -4*np.pi, 4*np.pi)) >>> f = lambda x, y: np.sin(np.sqrt(x ** 2 + y ** 2)) * x - y * np.cos(np.sqrt(x ** 2 + y ** 2)) >>> plot_resilience(f, ax=axes[2], extent=(-4*np.pi, 4*np.pi, -4*np.pi, 4*np.pi))

class nd2py.utils.plot.EqualizeNormalize(*args: Any, **kwargs: Any)

Bases: Normalize

按分布而非值进行归一化

__init__(samples, clip=False)

inverse(value)

nd2py.utils.plot.plotOD(ax, source: List[str], destination: List[str], flow: List[float], location: Dict[str, Tuple[float, float]], linetype: Literal['straight', 'parabola', 'rotated_parabola', 'projected_parabola'] = 'straight', N=100, zorder=10, **kwargs)

绘制OD流量

nd2py.utils.plot.clear_svg(path, debug=False)

matplotlib 生成的 svg 中会使用 <text style=”font: 9.8px ‘Arial’; text-anchor: middle” x=”80.307802” y=”193.900483”>2020-02-02</text> 的语法，而 Powerpoint 无法识别 font: 9.8px ‘Arial’; 的简写记法，只能识别 font-family: ‘Arial’; font-size: 9.8px; 的记法。因此需要进行转换。考虑的属性包括： - font-size - font-family - font-weight - font-style

nd2py.utils.plot.load_font()

• plt.title(“示例图表：数字平方”, fontproperties=font, size=15)

• plt.xlabel(“数字”, fontproperties=font, size=12)

• plt.ylabel(“平方”, fontproperties=font, size=12)

nd2py.utils.plot.adjust_text(texts, ax, step=0.01, max_iterations=100, mode='xy')

nd2py.utils.plot.use_chinese_font(fontpath='/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc')

nd2py.utils.plot.merge_axes(axes, fig=None)

nd2py.utils.render_markdown module

nd2py.utils.render_markdown.render_markdown(text: str, width=120, theme='staroffice') → str

nd2py.utils.render_python module

nd2py.utils.render_python.render_python(text: str, width=120, highlight_lines=[], theme='staroffice') → str

nd2py.utils.tag2ansi module

nd2py.utils.tag2ansi.tag2ansi(text: str) → str

将 [tag1 tag2] 转换为 ANSI 码并智能插入 reset。 支持多种颜色定义方式、样式组合、自动闭合。

nd2py.utils.timing module

Lightweight timing utilities for optional performance diagnostics.

class nd2py.utils.timing.Timer(unit='iter')

Bases: object

__init__(unit='iter')

计时器

add(n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace')

property time

property count

property pace

property speed

time_str()

count_str()

pace_str()

speed_str()

to_dict()

classmethod from_dict(dict)

class nd2py.utils.timing.NamedTimer(unit='iter')

Bases: Timer

__init__(unit='iter')

对 Timer 的扩展, 支持将 count 和 time 统计到不同的名称下

add(name, n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace', mode_of_detail: Literal['time', 'count', 'pace', 'speed'] | None = 'pace', mode_of_percent: Literal['by_time', 'by_count'] | None = 'by_time')

property names

property time

property count

get_named_time(name)

get_named_count(name)

get_named_pace(name)

get_named_speed(name)

property named_time

property named_count

property named_pace

property named_speed

class nd2py.utils.timing.ParallelTimer(unit='iter')

Bases: Timer

__init__(unit='iter')

对 Timer 的扩展, 支持将 count 统计到不同的名称下, 但 time 在各名称间共享

add(name, n=1, by: Literal['increment', 'absolute'] = 'increment')

clear(reset_last_add_time=False)

to_str(mode: Literal['time', 'count', 'pace', 'speed'] = 'pace', mode_of_detail: Literal['time', 'count', 'pace', 'speed'] | None = 'pace', mode_of_percent: Literal['by_time', 'by_count'] | None = 'by_time')

property names

property count

get_named_time(name)

get_named_count(name)

get_named_pace(name)

get_named_speed(name)

property named_time

property named_count

property named_pace

property named_speed

nd2py.utils.utils module

nd2py.utils.utils.softmax(x)

nd2py.utils.utils.seed_all(seed)