Spaces:
Sleeping
Sleeping
| from typing import Set, Text, List, Tuple, Dict | |
| from itertools import chain | |
| from .utils import get_project_root | |
| import numpy as np | |
| import json | |
| class HW_NATS_FastInterface: | |
| def __init__(self, | |
| datapath:str=str(get_project_root()) + "/data/nebuloss.json", | |
| indexpath:str=str(get_project_root()) + "/data/nats_arch_index.json", | |
| dataset:str="cifar10", | |
| device:Text="edgegpu", | |
| scores_sample_size:int=1e3): | |
| AVAILABLE_DATASETS = ["cifar10", "cifar100", "ImageNet16-120"] | |
| AVAILABLE_DEVICES = ["edgegpu", "eyeriss", "fpga"] | |
| # catch input errors | |
| if dataset not in AVAILABLE_DATASETS: | |
| raise ValueError(f"Dataset {dataset} not in {AVAILABLE_DATASETS}!") | |
| if device not in AVAILABLE_DEVICES and device is not None: | |
| raise ValueError(f"Device {device} not in {AVAILABLE_DEVICES}!") | |
| if isinstance(datapath, str): | |
| # parent init | |
| with open(datapath, "r") as datafile: | |
| self._data = { | |
| int(key): value for key, value in json.load(datafile).items() | |
| } | |
| elif isinstance(datapath, dict): | |
| self._data = { | |
| int(key): value for key, value in datapath.items() | |
| } | |
| else: | |
| raise ValueError(f"Datapath must be either a string or a dictionary, not {type(datapath)}") | |
| # importing the "/"-architecture <-> index from a json file | |
| with open(indexpath, "r") as indexfile: | |
| self._architecture_to_index = json.load(indexfile) | |
| # store dataset field | |
| self._dataset = dataset | |
| self.target_device = device | |
| # architectures to use to estimate mean and std for scores normalization | |
| self.random_indices = np.random.choice(len(self), int(scores_sample_size), replace=False) | |
| def __len__(self)->int: | |
| """Number of architectures in considered search space.""" | |
| return len(self._data) | |
| def __getitem__(self, idx:int) -> Dict: | |
| """Returns (untrained) network corresponding to index `idx`""" | |
| return self._data[idx] | |
| def __iter__(self): | |
| """Iterator method""" | |
| self.iteration_index = 0 | |
| return self | |
| def __next__(self): | |
| if self.iteration_index >= self.__len__(): | |
| raise StopIteration | |
| # access current element | |
| net = self[self.iteration_index] | |
| # update the iteration index | |
| self.iteration_index += 1 | |
| return net | |
| def data(self): | |
| return self._data | |
| def architecture_to_index(self): | |
| return self._architecture_to_index | |
| def name(self)->Text: | |
| return "nats" | |
| def ordered_all_ops(self)->List[Text]: | |
| """NASTS Bench available operations, ordered (without any precise logic)""" | |
| return ['skip_connect', 'nor_conv_1x1', 'nor_conv_3x3', 'none', 'avg_pool_3x3'] | |
| def architecture_len(self)->int: | |
| """Returns the number of different operations that uniquevoly define a given architecture""" | |
| return 6 | |
| def all_ops(self)->Set[Text]: | |
| """NASTS Bench available operations.""" | |
| return {'skip_connect', 'nor_conv_1x1', 'nor_conv_3x3', 'none', 'avg_pool_3x3'} | |
| def dataset(self)->Text: | |
| return self._dataset | |
| def change_dataset(self, new_dataset:Text)->None: | |
| """ | |
| Updates the current dataset with a new one. | |
| Raises ValueError when new_dataset is not one of ["cifar10", "cifar100", "imagenet16-120"] | |
| """ | |
| if new_dataset.lower() in self.NATS_datasets: | |
| self._dataset = new_dataset | |
| else: | |
| raise ValueError(f"New dataset {new_dataset} not in {self.NATS_datasets}") | |
| def get_score_mean(self, score_name:Text)->float: | |
| """ | |
| Calculate the mean score value across the dataset for the given score name. | |
| Args: | |
| score_name (Text): The name of the score for which to calculate the mean. | |
| Returns: | |
| float: The mean score value. | |
| Note: | |
| The score values are retrieved from each data point in the dataset and averaged. | |
| """ | |
| if not hasattr(self, f"mean_{score_name}"): | |
| # compute the mean on 1000 instances | |
| mean_score = np.mean([self[i][self.dataset][score_name] for i in self.random_indices]) | |
| # set the mean score accordingly | |
| setattr(self, f"mean_{score_name}", mean_score) | |
| self.get_score_mean(score_name=score_name) | |
| return getattr(self, f"mean_{score_name}") | |
| def get_score_std(self, score_name: Text) -> float: | |
| """ | |
| Calculate the standard deviation of the score values across the dataset for the given score name. | |
| Args: | |
| score_name (Text): The name of the score for which to calculate the standard deviation. | |
| Returns: | |
| float: The standard deviation of the score values. | |
| Note: | |
| The score values are retrieved from each data point in the dataset, and the standard deviation is calculated. | |
| """ | |
| if not hasattr(self, f"std_{score_name}"): | |
| # compute the mean on 1000 instances | |
| std_score = np.std([self[i][self.dataset][score_name] for i in self.random_indices]) | |
| # set the mean score accordingly | |
| setattr(self, f"std_{score_name}", std_score) | |
| self.get_score_std(score_name=score_name) | |
| return getattr(self, f"std_{score_name}") | |
| def generate_random_samples(self, n_samples:int=10)->Tuple[List[Text], List[int]]: | |
| """Generate a group of architectures chosen at random""" | |
| idxs = np.random.choice(self.__len__(), size=n_samples, replace=False) | |
| cell_structures = [self[i]["architecture_string"] for i in idxs] | |
| # return tinynets, cell_structures_string and the unique indices of the networks | |
| return cell_structures, idxs | |
| def list_to_architecture(self, input_list:List[str])->str: | |
| """ | |
| Reformats genotype as architecture string. | |
| This function clearly is specific for this very search space. | |
| """ | |
| return "|{}|+|{}|{}|+|{}|{}|{}|".format(*input_list) | |
| def architecture_to_list(self, architecture_string:Text)->List[Text]: | |
| """Turn architectures string into genotype list | |
| Args: | |
| architecture_string(str): String characterising the cell structure only. | |
| Returns: | |
| List[str]: List containing the operations in the input cell structure. | |
| In a genetic-algorithm setting, this description represents a genotype. | |
| """ | |
| # divide the input string into different levels | |
| subcells = architecture_string.split("+") | |
| # divide into different nodes to retrieve ops | |
| ops = chain(*[subcell.split("|")[1:-1] for subcell in subcells]) | |
| return list(ops) | |
| def list_to_accuracy(self, input_list:List[str])->float: | |
| """Returns the test accuracy of an input list representing the architecture. | |
| This list contains the operations. | |
| Args: | |
| input_list (List[str]): List of operations inside the architecture. | |
| Returns: | |
| float: Test accuracy (after 200 training epochs). | |
| """ | |
| # retrieving the index associated to this particular architecture | |
| arch_index = self.architecture_to_index["/".join(input_list)] | |
| return self[arch_index][self.dataset]["test_accuracy"] | |
| def architecture_to_accuracy(self, architecture_string:str)->float: | |
| """Returns the test accuracy of an architecture string. | |
| The architecture <-> index map is normalized to be as general as possible, hence some (minor) | |
| input processing is needed. | |
| Args: | |
| architecture_string (str): Architecture string. | |
| Returns: | |
| float: Test accuracy (after 200 training epochs). | |
| """ | |
| # retrieving the index associated to this particular architecture | |
| arch_index = self.architecture_to_index["/".join(self.architecture_to_list(architecture_string))] | |
| return self[arch_index][self.dataset]["test_accuracy"] | |
| def list_to_score(self, input_list:List[Text], score:Text)->float: | |
| """Returns the value of `score` of an input list representing the architecture. | |
| This list contains the operations. | |
| Args: | |
| input_list (List[Text]): List of operations inside the architecture. | |
| score (Text): Score of interest. | |
| Returns: | |
| float: Score value for `input_list`. | |
| """ | |
| arch_index = self.architecture_to_index["/".join(input_list)] | |
| return self[arch_index][self.dataset].get(score, None) | |
| def architecture_to_score(self, architecture_string:Text, score:Text)->float: | |
| """Returns the value of `score` of an architecture string. | |
| The architecture <-> index map is normalized to be as general as possible, hence some (minor) | |
| input processing is needed. | |
| Args: | |
| architecture_string (Text): Architecture string. | |
| score (Text): Score of interest. | |
| Returns: | |
| float: Score value for `architecture_string`. | |
| """ | |
| # retrieving the index associated to this particular architecture | |
| arch_index = self.architecture_to_index["/".join(self.architecture_to_list(architecture_string))] | |
| return self[arch_index][self.dataset].get(score, None) | |