5600673849

################################################################################
# Copyright (c) 2021 ContinualAI.                                              #
# Copyrights licensed under the MIT License.                                   #
# See the accompanying LICENSE file for terms.                                 #
#                                                                              #
# Date: 1-05-2020                                                              #
# Author(s): Vincenzo Lomonaco                                                 #
# E-mail: contact@continualai.org                                              #
# Website: www.continualai.org                                                 #
################################################################################

""" This module contains the high-level CORe50 benchmark generator. It
basically returns a iterable benchmark object ``GenericCLScenario`` given a
number of configuration parameters."""
from pathlib import Path
from typing import Union, Optional, Any

from torchvision.transforms import (
    ToTensor,
    Normalize,
    Compose,
    RandomHorizontalFlip,
)

from avalanche.benchmarks.classic.classic_benchmarks_utils import (
    check_vision_benchmark,
)
from avalanche.benchmarks.datasets import default_dataset_location
from avalanche.benchmarks.scenarios.generic_benchmark_creation import (
    create_generic_benchmark_from_filelists,
)
from avalanche.benchmarks.datasets.core50.core50 import CORe50Dataset

nbatch = {
    "ni": 8,
    "nc": 9,
    "nic": 79,
    "nicv2_79": 79,
    "nicv2_196": 196,
    "nicv2_391": 391,
}

scen2dirs = {
    "ni": "batches_filelists/NI_inc/",
    "nc": "batches_filelists/NC_inc/",
    "nic": "batches_filelists/NIC_inc/",
    "nicv2_79": "NIC_v2_79/",
    "nicv2_196": "NIC_v2_196/",
    "nicv2_391": "NIC_v2_391/",
}


normalize = Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])

_default_train_transform = Compose([ToTensor(), RandomHorizontalFlip(), normalize])

_default_eval_transform = Compose([ToTensor(), normalize])


def CORe50(
    *,
    scenario: str = "nicv2_391",
    run: int = 0,
    object_lvl: bool = True,
    mini: bool = False,
    train_transform: Optional[Any] = _default_train_transform,
    eval_transform: Optional[Any] = _default_eval_transform,
    dataset_root: Optional[Union[str, Path]] = None
):
    """
    Creates a CL benchmark for CORe50.

    If the dataset is not present in the computer, this method will
    automatically download and store it.

    This generator can be used to obtain the NI, NC, NIC and NICv2-* scenarios.

    The benchmark instance returned by this method will have two fields,
    `train_stream` and `test_stream`, which can be iterated to obtain
    training and test :class:`Experience`. Each Experience contains the
    `dataset` and the associated task label.

    The task label 0 will be assigned to each experience.

    The benchmark API is quite simple and is uniform across all benchmark
    generators. It is recommended to check the tutorial of the "benchmark" API,
    which contains usage examples ranging from "basic" to "advanced".

    :param scenario: CORe50 main scenario. It can be chosen between 'ni', 'nc',
        'nic', 'nicv2_79', 'nicv2_196' or 'nicv2_391.'
    :param run: number of run for the benchmark. Each run defines a different
        ordering. Must be a number between 0 and 9.
    :param object_lvl: True for a 50-way classification at the object level.
        False if you want to use the categories as classes. Default to True.
    :param mini: True for processing reduced 32x32 images instead of the
        original 128x128. Default to False.
    :param train_transform: The transformation to apply to the training data,
        e.g. a random crop, a normalization or a concatenation of different
        transformations (see torchvision.transform documentation for a
        comprehensive list of possible transformations). Defaults to None.
    :param eval_transform: The transformation to apply to the test data,
        e.g. a random crop, a normalization or a concatenation of different
        transformations (see torchvision.transform documentation for a
        comprehensive list of possible transformations). Defaults to None.
    :param dataset_root: Absolute path indicating where to store the dataset
        and related metadata. Defaults to None, which means that the default
        location for
        'core50' will be used.

    :returns: a properly initialized :class:`GenericCLScenario` instance.
    """

    assert 0 <= run <= 9, (
        "Pre-defined run of CORe50 are only 10. Indicate " "a number between 0 and 9."
    )
    assert scenario in nbatch.keys(), (
        "The selected scenario is note "
        "recognized: it should be 'ni', 'nc',"
        "'nic', 'nicv2_79', 'nicv2_196' or "
        "'nicv2_391'."
    )

    if dataset_root is None:
        dataset_root = default_dataset_location("core50")

    # Download the dataset and initialize filelists
    core_data = CORe50Dataset(root=dataset_root, mini=mini)

    root = core_data.root
    if mini:
        bp = "core50_32x32"
    else:
        bp = "core50_128x128"
    root_img = root / bp

    if object_lvl:
        suffix = "/"
    else:
        suffix = "_cat/"
    filelists_bp = scen2dirs[scenario][:-1] + suffix + "run" + str(run)
    train_failists_paths = []
    for batch_id in range(nbatch[scenario]):
        train_failists_paths.append(
            root
            / filelists_bp
            / ("train_batch_" + str(batch_id).zfill(2) + "_filelist.txt")
        )

    benchmark_obj = create_generic_benchmark_from_filelists(
        root_img,
        train_failists_paths,
        [root / filelists_bp / "test_filelist.txt"],
        task_labels=[0 for _ in range(nbatch[scenario])],
        complete_test_set_only=True,
        train_transform=train_transform,
        eval_transform=eval_transform,
    )

    if scenario == "nc":
        n_classes_per_exp = []
        classes_order = []
        for exp in benchmark_obj.train_stream:
            exp_dataset = exp.dataset
            unique_targets = list(
                sorted(set(int(x) for x in exp_dataset.targets))  # type: ignore
            )
            n_classes_per_exp.append(len(unique_targets))
            classes_order.extend(unique_targets)
        setattr(benchmark_obj, "n_classes_per_exp", n_classes_per_exp)
        setattr(benchmark_obj, "classes_order", classes_order)
    setattr(benchmark_obj, "n_classes", 50 if object_lvl else 10)

    return benchmark_obj


__all__ = ["CORe50"]

if __name__ == "__main__":
    import sys

    benchmark_instance = CORe50(scenario="nicv2_79", mini=False)
    check_vision_benchmark(benchmark_instance)
    sys.exit(0)

1	################################################################################
2	# Copyright (c) 2021 ContinualAI. #
3	# Copyrights licensed under the MIT License. #
4	# See the accompanying LICENSE file for terms. #
5	# #
6	# Date: 1-05-2020 #
7	# Author(s): Vincenzo Lomonaco #
8	# E-mail: contact@continualai.org #
9	# Website: www.continualai.org #
10	################################################################################
11
12	""" This module contains the high-level CORe50 benchmark generator. It	4✔
13	basically returns a iterable benchmark object ``GenericCLScenario`` given a
14	number of configuration parameters."""
15	from pathlib import Path	4✔
16	from typing import Union, Optional, Any	4✔
17
18	from torchvision.transforms import (	4✔
19	ToTensor,
20	Normalize,
21	Compose,
22	RandomHorizontalFlip,
23	)
24
25	from avalanche.benchmarks.classic.classic_benchmarks_utils import (	4✔
26	check_vision_benchmark,
27	)
28	from avalanche.benchmarks.datasets import default_dataset_location	4✔
29	from avalanche.benchmarks.scenarios.generic_benchmark_creation import (	4✔
30	create_generic_benchmark_from_filelists,
31	)
32	from avalanche.benchmarks.datasets.core50.core50 import CORe50Dataset	4✔
33
34	nbatch = {	4✔
35	"ni": 8,
36	"nc": 9,
37	"nic": 79,
38	"nicv2_79": 79,
39	"nicv2_196": 196,
40	"nicv2_391": 391,
41	}
42
43	scen2dirs = {	4✔
44	"ni": "batches_filelists/NI_inc/",
45	"nc": "batches_filelists/NC_inc/",
46	"nic": "batches_filelists/NIC_inc/",
47	"nicv2_79": "NIC_v2_79/",
48	"nicv2_196": "NIC_v2_196/",
49	"nicv2_391": "NIC_v2_391/",
50	}
51
52
53	normalize = Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])	4✔
54
55	_default_train_transform = Compose([ToTensor(), RandomHorizontalFlip(), normalize])	4✔
56
57	_default_eval_transform = Compose([ToTensor(), normalize])	4✔
58
59
60	def CORe50(	4✔
61	*,
62	scenario: str = "nicv2_391",
63	run: int = 0,
64	object_lvl: bool = True,
65	mini: bool = False,
66	train_transform: Optional[Any] = _default_train_transform,
67	eval_transform: Optional[Any] = _default_eval_transform,
68	dataset_root: Optional[Union[str, Path]] = None
69	):
70	"""
71	Creates a CL benchmark for CORe50.
72
73	If the dataset is not present in the computer, this method will
74	automatically download and store it.
75
76	This generator can be used to obtain the NI, NC, NIC and NICv2-* scenarios.
77
78	The benchmark instance returned by this method will have two fields,
79	`train_stream` and `test_stream`, which can be iterated to obtain
80	training and test :class:`Experience`. Each Experience contains the
81	`dataset` and the associated task label.
82
83	The task label 0 will be assigned to each experience.
84
85	The benchmark API is quite simple and is uniform across all benchmark
86	generators. It is recommended to check the tutorial of the "benchmark" API,
87	which contains usage examples ranging from "basic" to "advanced".
88
89	:param scenario: CORe50 main scenario. It can be chosen between 'ni', 'nc',
90	'nic', 'nicv2_79', 'nicv2_196' or 'nicv2_391.'
91	:param run: number of run for the benchmark. Each run defines a different
92	ordering. Must be a number between 0 and 9.
93	:param object_lvl: True for a 50-way classification at the object level.
94	False if you want to use the categories as classes. Default to True.
95	:param mini: True for processing reduced 32x32 images instead of the
96	original 128x128. Default to False.
97	:param train_transform: The transformation to apply to the training data,
98	e.g. a random crop, a normalization or a concatenation of different
99	transformations (see torchvision.transform documentation for a
100	comprehensive list of possible transformations). Defaults to None.
101	:param eval_transform: The transformation to apply to the test data,
102	e.g. a random crop, a normalization or a concatenation of different
103	transformations (see torchvision.transform documentation for a
104	comprehensive list of possible transformations). Defaults to None.
105	:param dataset_root: Absolute path indicating where to store the dataset
106	and related metadata. Defaults to None, which means that the default
107	location for
108	'core50' will be used.
109
110	:returns: a properly initialized :class:`GenericCLScenario` instance.
111	"""
112
113	assert 0 <= run <= 9, (	×
114	"Pre-defined run of CORe50 are only 10. Indicate " "a number between 0 and 9."
115	)
116	assert scenario in nbatch.keys(), (	×
117	"The selected scenario is note "
118	"recognized: it should be 'ni', 'nc',"
119	"'nic', 'nicv2_79', 'nicv2_196' or "
120	"'nicv2_391'."
121	)
122
123	if dataset_root is None:	×
124	dataset_root = default_dataset_location("core50")	×
125
126	# Download the dataset and initialize filelists
127	core_data = CORe50Dataset(root=dataset_root, mini=mini)	×
128
129	root = core_data.root	×
130	if mini:	×
131	bp = "core50_32x32"	×
132	else:
133	bp = "core50_128x128"	×
134	root_img = root / bp	×
135
136	if object_lvl:	×
137	suffix = "/"	×
138	else:
139	suffix = "_cat/"	×
140	filelists_bp = scen2dirs[scenario][:-1] + suffix + "run" + str(run)	×
141	train_failists_paths = []	×
142	for batch_id in range(nbatch[scenario]):	×
143	train_failists_paths.append(	×
144	root
145	/ filelists_bp
146	/ ("train_batch_" + str(batch_id).zfill(2) + "_filelist.txt")
147	)
148
149	benchmark_obj = create_generic_benchmark_from_filelists(	×
150	root_img,
151	train_failists_paths,
152	[root / filelists_bp / "test_filelist.txt"],
153	task_labels=[0 for _ in range(nbatch[scenario])],
154	complete_test_set_only=True,
155	train_transform=train_transform,
156	eval_transform=eval_transform,
157	)
158
NEW 159	if scenario == "nc":	×
NEW 160	n_classes_per_exp = []	×
NEW 161	classes_order = []	×
NEW 162	for exp in benchmark_obj.train_stream:	×
NEW 163	exp_dataset = exp.dataset	×
NEW 164	unique_targets = list(	×
165	sorted(set(int(x) for x in exp_dataset.targets)) # type: ignore
166	)
NEW 167	n_classes_per_exp.append(len(unique_targets))	×
NEW 168	classes_order.extend(unique_targets)	×
NEW 169	setattr(benchmark_obj, "n_classes_per_exp", n_classes_per_exp)	×
NEW 170	setattr(benchmark_obj, "classes_order", classes_order)	×
NEW 171	setattr(benchmark_obj, "n_classes", 50 if object_lvl else 10)	×
172
UNCOV 173	return benchmark_obj	×
174
175
176	__all__ = ["CORe50"]	4✔
177
178	if __name__ == "__main__":	4✔
179	import sys	×
180
181	benchmark_instance = CORe50(scenario="nicv2_79", mini=False)	×
182	check_vision_benchmark(benchmark_instance)	×
183	sys.exit(0)	×

ContinualAI / avalanche / 5600673849

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous