spikingjelly.datasets.asl_dvs 源代码

import os
from pathlib import Path
from typing import Callable, Optional, Tuple, Union
import multiprocessing
from concurrent.futures import ThreadPoolExecutor
import time
import shutil

import scipy.io
from torchvision.datasets.utils import extract_archive

from .. import configure
from .base import NeuromorphicDatasetFolder
from . import utils


__all__ = ["ASLDVS"]


def _read_mat_save_to_np(mat_file: Union[str, Path], np_file: Union[str, Path]):
    mat_file, np_file = str(mat_file), str(np_file)
    events = scipy.io.loadmat(mat_file)
    events = {
        "t": events["ts"].squeeze(),
        "x": 239 - events["x"].squeeze(),
        "y": 179 - events["y"].squeeze(),
        "p": events["pol"].squeeze(),
    }
    utils.np_savez(np_file, t=events["t"], x=events["x"], y=events["y"], p=events["p"])
    print(f"Save [{mat_file}] to [{np_file}].")


[文档] class ASLDVS(NeuromorphicDatasetFolder): def __init__( self, root: str, data_type: str = "event", frames_number: int = None, split_by: str = None, duration: int = None, custom_integrate_function: Callable = None, custom_integrated_frames_dir_name: str = None, transform: Optional[Callable] = None, target_transform: Optional[Callable] = None, ): """ **API Language:** :ref:`中文 <ASLDVS.__init__-cn>` | :ref:`English <ASLDVS.__init__-en>` ---- .. _ASLDVS.__init__-cn: * **中文** ASL-DVS 数据集,由 `Graph-based Object Classification for Neuromorphic Vision Sensing <https://openaccess.thecvf.com/content_ICCV_2019/html/Bi_Graph-Based_Object_Classification_for_Neuromorphic_Vision_Sensing_ICCV_2019_paper.html>`_ 提出。 有关参数信息的更多详细信息,请参考 :class:`NeuromorphicDatasetFolder <spikingjelly.datasets.base.NeuromorphicDatasetFolder>` .. note:: ASLDVS 的 Dropbox 链接已过期。用户可以通过以下命令从 OpenI 镜像手动下载此数据集: .. code:: shell pip install openi openi dataset download OpenI/ASLDVS --local_dir ./ASLDVS --max_workers 10 然后您可以解压 ``ASLDVS.zip`` 并获取 ``ICCV2019_DVS_dataset.zip`` 。 ---- .. _ASLDVS.__init__-en: * **English** The ASL-DVS dataset, which is proposed by `Graph-based Object Classification for Neuromorphic Vision Sensing <https://openaccess.thecvf.com/content_ICCV_2019/html/Bi_Graph-Based_Object_Classification_for_Neuromorphic_Vision_Sensing_ICCV_2019_paper.html>`_. Refer to :class:`NeuromorphicDatasetFolder <spikingjelly.datasets.base.NeuromorphicDatasetFolder>` for more details about params information. .. note:: ASLDVS's Dropbox link is expired. Users can download this dataset from the OpenI mirror manually by the following commands: .. code:: shell pip install openi openi dataset download OpenI/ASLDVS --local_dir ./ASLDVS --max_workers 10 Then you can extract ``ASLDVS.zip`` and get ``ICCV2019_DVS_dataset.zip`` . :param root: 数据集的根路径 :type root: Union[str, Path] :param data_type: ``\"event\"`` 或 ``\"frame\"`` :type data_type: str :param frames_number: 积分帧的数量 :type frames_number: Optional[int] :param split_by: ``\"time\"`` 或 ``\"number\"`` :type split_by: Optional[str] :param duration: 每帧的时间时长 :type duration: Optional[int] :param custom_integrate_function: 用户自定义积分函数 :type custom_integrate_function: Optional[Callable] :param custom_integrated_frames_dir_name: 自定义积分帧目录名 :type custom_integrated_frames_dir_name: Optional[str] :param transform: 数据变换 :type transform: Optional[Callable] :param target_transform: 标签变换 :type target_transform: Optional[Callable] :param root: Root directory of the dataset :type root: Union[str, Path] :param data_type: ``\"event\"`` or ``\"frame\"`` :type data_type: str :param frames_number: Number of frames to integrate :type frames_number: Optional[int] :param split_by: ``\"time\"`` or ``\"number\"`` :type split_by: Optional[str] :param duration: Time duration per frame :type duration: Optional[int] :param custom_integrate_function: User-defined integrate function :type custom_integrate_function: Optional[Callable] :param custom_integrated_frames_dir_name: Custom frames directory name :type custom_integrated_frames_dir_name: Optional[str] :param transform: Transform function :type transform: Optional[Callable] :param target_transform: Target transform function :type target_transform: Optional[Callable] :return: None :rtype: None """ super().__init__( root, None, data_type, frames_number, split_by, duration, custom_integrate_function, custom_integrated_frames_dir_name, transform, target_transform, )
[文档] @classmethod def get_H_W(cls) -> Tuple: r""" **API Language:** :ref:`中文 <asl_dvs.get_H_W-cn>` | :ref:`English <asl_dvs.get_H_W-en>` ---- .. _asl_dvs.get_H_W-cn: * **中文** :return: ``(180, 240)`` :rtype: Tuple ---- .. _asl_dvs.get_H_W-en: * **English** :return: ``(180, 240)`` :rtype: Tuple """ return 180, 240
[文档] @classmethod def resource_url_md5(cls) -> list: print( "The ICCV2019_DVS_dataset.zip is packed by dropbox. We find that the" "MD5 of this zip file can change. So, MD5 check will not be used for" "ASL-DVS dataset." ) print( "Update: The Dropbox link is expired now. You can download this dataset" "from the OpenI mirror manually by the following commands:\n" "----------\n" "pip install openi\n" "openi dataset download OpenI/ASLDVS --local_dir ./ASLDVS --max_workers 10\n" "----------\n" 'Then you can extract "ASLDVS.zip" and get "ICCV2019_DVS_dataset.zip".' ) url = ( "https://www.dropbox.com/sh/ibq0jsicatn7l6r/AACNrNELV56rs1YInMWUs9CAa?dl=0" ) return [("ICCV2019_DVS_dataset.zip", url, None)]
[文档] @classmethod def downloadable(cls) -> bool: """ :return: ``False`` :rtype: bool """ return False
[文档] @classmethod def extract_downloaded_files(cls, download_root: Path, extract_root: Path): temp_ext_dir = download_root / "temp_ext" temp_ext_dir.mkdir() print(f"Mkdir [{temp_ext_dir}].") extract_archive(download_root / "ICCV2019_DVS_dataset.zip", temp_ext_dir) with ThreadPoolExecutor(max_workers=min(multiprocessing.cpu_count(), 2)) as tpe: futures = [] for zip_file in temp_ext_dir.iterdir(): if zip_file.suffix == ".zip": print(f"Extract [{zip_file}] to [{extract_root}].") futures.append(tpe.submit(extract_archive, zip_file, extract_root)) for future in futures: future.result() shutil.rmtree(temp_ext_dir) print(f"Rmtree [{temp_ext_dir}].")
[文档] @classmethod def create_raw_from_extracted(cls, extract_root: Path, raw_root: Path): t_ckp = time.time() with ThreadPoolExecutor( max_workers=min( multiprocessing.cpu_count(), configure.max_threads_number_for_datasets_preprocess, ) ) as tpe: futures = [] for class_name in os.listdir(extract_root): mat_dir = extract_root / class_name np_dir = raw_root / class_name np_dir.mkdir() print(f"Mkdir [{np_dir}].") for bin_file in os.listdir(mat_dir): source_file = mat_dir / bin_file target_file = np_dir / (os.path.splitext(bin_file)[0] + ".npz") print(f"Start to convert [{source_file}] to [{target_file}].") futures.append( tpe.submit(_read_mat_save_to_np, source_file, target_file) ) for future in futures: future.result() print(f"Used time = [{round(time.time() - t_ckp, 2)}s].")