spikingjelly.activation_based.monitor package

Module contents

spikingjelly.activation_based.monitor.unpack_len1_tuple(x: tuple)[源代码]
class spikingjelly.activation_based.monitor.BaseMonitor[源代码]

基类:object

clear_recorded_data()[源代码]
enable()[源代码]
disable()[源代码]
is_enable()[源代码]
remove_hooks()[源代码]
class spikingjelly.activation_based.monitor.OutputMonitor(net: ~torch.nn.modules.module.Module, instance: ~typing.Optional[~typing.Any] = None, function_on_output: ~typing.Callable = <function OutputMonitor.<lambda>>)[源代码]

基类:BaseMonitor

参数:
  • net (nn.Module) – 一个神经网络

  • instance (Any or tuple) – 被监视的模块的数据类型。若为 None 则表示类型为 type(net)

  • function_on_output (Callable) – 作用于被监控的模块输出的自定义的函数

net 中所有类型为 instance 的模块的输出使用 function_on_output 作用后,记录到类型为 list`self.records 中。 可以通过 self.enable()self.disable() 来启用或停用这个监视器。 可以通过 self.clear_recorded_data() 来清除已经记录的数据。

阅读监视器的教程以获得更多信息。

示例代码:

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.OutputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([[0., 0., 0., 1.]]), tensor([[0., 0.]])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([[0., 0., 0., 1.]])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([[0., 0., 0., 1.]])]
参数:
  • net (nn.Module) – a network

  • instance (Any or tuple) – the instance of modules to be monitored. If None, it will be regarded as type(net)

  • function_on_output (Callable) – the function that applies on the monitored modules’ outputs

Applies function_on_output on outputs of all modules whose instances are instance in net, and records the data into self.records, which is a list. Call self.enable() or self.disable() to enable or disable the monitor. Call self.clear_recorded_data() to clear the recorded data.

Refer to the tutorial about the monitor for more details.

Codes example:

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.OutputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([[0., 0., 0., 1.]]), tensor([[0., 0.]])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([[0., 0., 0., 1.]])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([[0., 0., 0., 1.]])]
create_hook(name)[源代码]
class spikingjelly.activation_based.monitor.InputMonitor(net: ~torch.nn.modules.module.Module, instance: ~typing.Optional[~typing.Any] = None, function_on_input: ~typing.Callable = <function InputMonitor.<lambda>>)[源代码]

基类:BaseMonitor

参数:
  • net (nn.Module) – 一个神经网络

  • instance (Any or tuple) – 被监视的模块的数据类型。若为 None 则表示类型为 type(net)

  • function_on_input (Callable) – 作用于被监控的模块输入的自定义的函数

net 中所有类型为 instance 的模块的输入使用 function_on_input 作用后,记录到类型为 list`self.records 中。 可以通过 self.enable()self.disable() 来启用或停用这个监视器。 可以通过 self.clear_recorded_data() 来清除已经记录的数据。

阅读监视器的教程以获得更多信息。

示例代码:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.InputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([[1.0165, 1.1934, 0.9347, 0.9539]]), tensor([[0.9115, 0.9508]])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([[1.0165, 1.1934, 0.9347, 0.9539]])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([[1.0165, 1.1934, 0.9347, 0.9539]])]
参数:
  • net (nn.Module) – a network

  • instance (Any or tuple) – the instance of modules to be monitored. If None, it will be regarded as type(net)

  • function_on_input (Callable) – the function that applies on the monitored modules’ inputs

Applies function_on_input on inputs of all modules whose instances are instance in net, and records the data into self.records, which is a list. Call self.enable() or self.disable() to enable or disable the monitor. Call self.clear_recorded_data() to clear the recorded data.

Refer to the tutorial about the monitor for more details.

Codes example:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.InputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([[1.0165, 1.1934, 0.9347, 0.9539]]), tensor([[0.9115, 0.9508]])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([[1.0165, 1.1934, 0.9347, 0.9539]])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([[1.0165, 1.1934, 0.9347, 0.9539]])]
create_hook(name)[源代码]
class spikingjelly.activation_based.monitor.AttributeMonitor(attribute_name: str, pre_forward: bool, net: ~torch.nn.modules.module.Module, instance: ~typing.Optional[~typing.Any] = None, function_on_attribute: ~typing.Callable = <function AttributeMonitor.<lambda>>)[源代码]

基类:BaseMonitor

参数:
  • attribute_name (str) – 要监控的成员变量的名字

  • pre_forward (bool) – 若为 True,则记录模块在完成前向传播前的成员变量,否则记录完成前向传播后的变量

  • net (nn.Module) – 一个神经网络

  • instance (Any or tuple) – 被监视的模块的数据类型。若为 None 则表示类型为 type(net)

  • function_on_attribute (Callable) – 作用于被监控的模块 m 的成员 m.attribute_name 的自定义的函数

net 中所有类型为 instance 的模块 m 的成员 m.attribute_name 使用 function_on_attribute 作用后,记录到类型为 list`self.records。 可以通过 self.enable()self.disable() 来启用或停用这个监视器。 可以通过 self.clear_recorded_data() 来清除已经记录的数据。

阅读监视器的教程以获得更多信息。

示例代码:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.AttributeMonitor('v', False, net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([0.0000, 0.6854, 0.0000, 0.7968]), tensor([0.4472, 0.0000])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([0.0000, 0.6854, 0.0000, 0.7968])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([0.0000, 0.6854, 0.0000, 0.7968])]
参数:
  • attribute_name (str) – the monitored attribute’s name

  • pre_forward (bool) – If True, recording the attribute before forward, otherwise recording the attribute after forward

  • net (nn.Module) – a network

  • instance (Any or tuple) – the instance of modules to be monitored. If None, it will be regarded as type(net)

  • function_on_attribute (Callable) – the function that applies on each monitored module’s attribute

Applies function_on_attribute on m.attribute_name of each monitored module m whose instance is instance in net, and records the data into self.records, which is a list. Call self.enable() or self.disable() to enable or disable the monitor. Call self.clear_recorded_data() to clear the recorded data.

Refer to the tutorial about the monitor for more details.

Codes example:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.AttributeMonitor('v', False, net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([0.0000, 0.6854, 0.0000, 0.7968]), tensor([0.4472, 0.0000])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([0.0000, 0.6854, 0.0000, 0.7968])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([0.0000, 0.6854, 0.0000, 0.7968])]
create_hook(name)[源代码]
class spikingjelly.activation_based.monitor.GradInputMonitor(net: ~torch.nn.modules.module.Module, instance: ~typing.Optional[~typing.Any] = None, function_on_grad_input: ~typing.Callable = <function GradInputMonitor.<lambda>>)[源代码]

基类:BaseMonitor

参数:
  • net (nn.Module) – 一个神经网络

  • instance (Any or tuple) – 被监视的模块的数据类型。若为 None 则表示类型为 type(net)

  • function_on_grad_input (Callable) – 作用于被监控的模块输出的输入的梯度的函数

net 中所有类型为 instance 的模块的输入的梯度使用 function_on_grad_input 作用后,记录到类型为 list`self.records 中。 可以通过 self.enable()self.disable() 来启用或停用这个监视器。 可以通过 self.clear_recorded_data() 来清除已经记录的数据。

阅读监视器的教程以获得更多信息。

备注

对于一个模块,输入为 \(X\),输出为 \(Y\),损失为 \(L\),则 GradInputMonitor 记录的是对输入的梯度 \(\frac{\partial L}{\partial X}\)

示例代码:

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.GradInputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([0.0000, 0.6854, 0.0000, 0.7968]), tensor([0.4472, 0.0000])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([0.0000, 0.6854, 0.0000, 0.7968])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([0.0000, 0.6854, 0.0000, 0.7968])]
参数:
  • net (nn.Module) – a network

  • instance (Any or tuple) – the instance of modules to be monitored. If None, it will be regarded as type(net)

  • function_on_grad_input (Callable) – the function that applies on the grad of monitored modules’ inputs

Applies function_on_grad_input on grad of inputs of all modules whose instances are instance in net, and records the data into self.records, which is a list. Call self.enable() or self.disable() to enable or disable the monitor. Call self.clear_recorded_data() to clear the recorded data.

Refer to the tutorial about the monitor for more details.

Note

Denote the input and output of the monitored module as \(X\) and \(Y\), and the loss is \(L\), then GradInputMonitor will record the gradient of input, which is \(\frac{\partial L}{\partial X}\).

Codes example:

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.GradInputMonitor(net, instance=neuron.IFNode)

with torch.no_grad():
    y = net(torch.rand([1, 8]))
    print(f'mtor.records={mtor.records}')
    # mtor.records=[tensor([0.0000, 0.6854, 0.0000, 0.7968]), tensor([0.4472, 0.0000])]
    print(f'mtor[0]={mtor[0]}')
    # mtor[0]=tensor([0.0000, 0.6854, 0.0000, 0.7968])
    print(f'mtor.monitored_layers={mtor.monitored_layers}')
    # mtor.monitored_layers=['sn1', 'sn2']
    print(f"mtor['sn1']={mtor['sn1']}")
    # mtor['sn1']=[tensor([0.0000, 0.6854, 0.0000, 0.7968])]
create_hook(name)[源代码]
class spikingjelly.activation_based.monitor.GradOutputMonitor(net: ~torch.nn.modules.module.Module, instance: ~typing.Optional[~typing.Any] = None, function_on_grad_output: ~typing.Callable = <function GradOutputMonitor.<lambda>>)[源代码]

基类:BaseMonitor

参数:
  • net (nn.Module) – 一个神经网络

  • instance (Any or tuple) – 被监视的模块的数据类型。若为 None 则表示类型为 type(net)

  • function_on_grad_output (Callable) – 作用于被监控的模块输出的输出的的梯度的函数

net 中所有类型为 instance 的模块的输出的梯度使用 function_on_grad_output 作用后,记录到类型为 list`self.records 中。 可以通过 self.enable()self.disable() 来启用或停用这个监视器。 可以通过 self.clear_recorded_data() 来清除已经记录的数据。

阅读监视器的教程以获得更多信息。

备注

对于一个模块,输入为 \(X\),输出为 \(Y\),损失为 \(L\),则 GradOutputMonitor 记录的是对输出的梯度 \(\frac{\partial L}{\partial Y}\)

示例代码:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.GradOutputMonitor(net, instance=neuron.IFNode)

net(torch.rand([1, 8])).sum().backward()
print(f'mtor.records={mtor.records}')
# mtor.records=[tensor([[1., 1.]]), tensor([[0.1372, 0.1081, 0.0880, 0.1089]])]
print(f'mtor[0]={mtor[0]}')
# mtor[0]=tensor([[1., 1.]])
print(f'mtor.monitored_layers={mtor.monitored_layers}')
# mtor.monitored_layers=['sn1', 'sn2']
print(f"mtor['sn1']={mtor['sn1']}")
# mtor['sn1']=[tensor([[0.1372, 0.1081, 0.0880, 0.1089]])]
参数:
  • net (nn.Module) – a network

  • instance (Any or tuple) – the instance of modules to be monitored. If None, it will be regarded as type(net)

  • function_on_grad_output (Callable) – the function that applies on the grad of monitored modules’ inputs

Applies function_on_grad_output on grad of outputs of all modules whose instances are instance in net, and records the data into self.records, which is a list. Call self.enable() or self.disable() to enable or disable the monitor. Call self.clear_recorded_data() to clear the recorded data.

Refer to the tutorial about the monitor for more details.

Note

Denote the input and output of the monitored module as \(X\) and \(Y\), and the loss is \(L\), then GradOutputMonitor will record the gradient of output, which is \(\frac{\partial L}{\partial Y}\).

Codes example:

import torch
import torch.nn as nn
from spikingjelly.activation_based import monitor, neuron, functional, layer

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = layer.Linear(8, 4)
        self.sn1 = neuron.IFNode()
        self.fc2 = layer.Linear(4, 2)
        self.sn2 = neuron.IFNode()
        functional.set_step_mode(self, 'm')

    def forward(self, x_seq: torch.Tensor):
        x_seq = self.fc1(x_seq)
        x_seq = self.sn1(x_seq)
        x_seq = self.fc2(x_seq)
        x_seq = self.sn2(x_seq)
        return x_seq

net = Net()
for param in net.parameters():
    param.data.abs_()

mtor = monitor.GradOutputMonitor(net, instance=neuron.IFNode)

net(torch.rand([1, 8])).sum().backward()
print(f'mtor.records={mtor.records}')
# mtor.records=[tensor([[1., 1.]]), tensor([[0.1372, 0.1081, 0.0880, 0.1089]])]
print(f'mtor[0]={mtor[0]}')
# mtor[0]=tensor([[1., 1.]])
print(f'mtor.monitored_layers={mtor.monitored_layers}')
# mtor.monitored_layers=['sn1', 'sn2']
print(f"mtor['sn1']={mtor['sn1']}")
# mtor['sn1']=[tensor([[0.1372, 0.1081, 0.0880, 0.1089]])]
create_hook(name)[源代码]
class spikingjelly.activation_based.monitor.GPUMonitor(log_dir: Optional[str] = None, gpu_ids: tuple = (0,), interval: float = 600.0, start_now=True)[源代码]

基类:Thread

参数:
  • log_dir (str) – 使用 tensorboard 保存GPU数据的文件夹. 若为None,则日志不会保存,而是直接 print

  • gpu_ids (tuple) – 监视的GPU,例如 (0, 1, 2, 3)。默认为 (0, )

  • interval (float) – 记录数据的间隔,单位是秒

  • start_now – 若为 True 则初始化后会立刻开始记录数据,否则需要手动调用 start() 后才开始记录数据

GPU监视器,可以开启一个新的线程来记录 gpu_ids 的使用率和显存使用情况,每 interval 秒记录一次数据。

警告

在主线程的工作完成后一定要调用GPU监视器的 stop() 函数,否则主线程不会退出。

Codes example:

import time

gm = GPUMonitor(interval=1)
time.sleep(2)  # make the main thread sleep
gm.stop()

# The outputs are:

# 2022-04-28 10:52:25
# utilization.gpu [%], memory.used [MiB]
# 0 %, 376 MiB
参数:
  • log_dir (str) – the directory for saving logs with tensorboard. If it is None, this module will print logs

  • gpu_ids (tuple) – the id of GPUs to be monitored, e.g., (0, 1, 2, 3). The default value is (0, )

  • interval (float) – the recording interval (in seconds)

  • start_now – if true, the monitor will start to record now. Otherwise, it will start after the user call start() manually

The GPU monitor, which starts a new thread to record the utilization and memory used of gpu_ids every interval seconds.

Warning

Do not forget to call this module’s stop() after the main thread finishes its job, otherwise the main thread will never stop!

Codes example:

import time

gm = GPUMonitor(interval=1)
time.sleep(2)  # make the main thread sleep
gm.stop()

# The outputs are:

# 2022-04-28 10:52:25
# utilization.gpu [%], memory.used [MiB]
# 0 %, 376 MiB
stop()[源代码]
run()[源代码]