Source code for neurio.tasks.classification.imageNetClassification

#!/user/bin/env python

"""
Author: Romain Gaulier
Email: romain.gaulier@csem.ch
Copyright: CSEM, 2023
Creation: 26.04.23
Description: Pipeline for MNIST dataset
"""


import numpy as np
import tensorflow as tf
from pycocotools.coco import COCO

import sys
import os
import subprocess
import wget
from pycocotools.cocoeval import COCOeval

from neurio.tasks.task import Task


[docs]class ImageNetClassification(Task): def __init__(self): super().__init__() self.x = None self.y = None project_path = os.getcwd() # Download ImageNet dataset url = 'http://image-net.org/image/ILSVRC2017/ILSVRC2017_CLS-LOC.tar.gz' save_dir = 'datasets' save_filename = 'ILSVRC2017_CLS-LOC.tar.gz' save_path = os.path.join(project_path, save_dir, save_filename) wget.download(url, save_path) # Load dataset image_size = (224, 224) # example to be discussed batch_size = 32 # example to be discussed self.dataset = tf.keras.preprocessing.image_dataset_from_directory( save_path, image_size=image_size, batch_size=batch_size, label_mode="categorical") # Download COCO coco_api_dir = os.path.join(project_path, 'cocoApi') subprocess.call(["git", "clone", "https://github.com/cocodataset/cocoapi.git", coco_api_dir]) # Run setup.py script of COCO API setup_script = os.path.join(coco_api_dir, "PYTHONAPI", "setup.py") subprocess.call(["python", setup_script, "build_ext", "--inplace"]) # Download COCO annotations annotations_url = "http://images.cocodataset.org/annotations/annotations_trainval2017.zip" annotations_dir = os.path.join(project_path, 'annotations') subprocess.call(["wget", annotations_url, "-P", annotations_dir]) annotations_zip = os.path.join(annotations_dir, "annotations_trainval2017.zip") subprocess.call(["unzip", annotations_zip, "-d", annotations_dir]) self.coco_gt = COCO(annotations_dir + 'instances_val2017.json') coco_eval = COCOeval(self.coco_gt) self.metric = coco_eval self.prepare_data()
[docs] def preprocess(self): """ Preprocess the data :return: """ (x_train, y_train), (x_test, y_test) = self.dataset.load_data() x_train = np.repeat(np.expand_dims(x_train, axis=3), 3, axis=3) x_test = np.repeat(np.expand_dims(x_test, axis=3), 3, axis=3) y_train = tf.keras.utils.to_categorical(y_train) y_test = tf.keras.utils.to_categorical(y_test) # Normalize value x_train = (x_train / 255.0).astype(np.float32) x_test = (x_test / 255.0).astype(np.float32) self.x = x_train.numpy() # Not sure? self.y = y_train.numpy()
[docs] def evaluate(self, y_train, y_pred): self.metric.cocoDt = self.coco_gt.loadRes(y_pred) self.metric.evaluate() self.metric.accumulate() self.metric.summarize() mAP = self.metric.stats[0] return mAP.numpy()