Added pipeline

fcnutr.py

@@ -0,0 +1,38 @@
+from tensorflow import keras
+
+class FCNutr(keras.Model):
+    def __init__(self, nutr_names, crop_size, hidden_dim, num_shared_hidden, use_dropout):
+        super().__init__()
+        
+        self.cnn = keras.applications.InceptionV3(
+            include_top=False, 
+            weights="imagenet", 
+            input_shape=crop_size + (3,), 
+            pooling='avg'
+        )
+
+        shared_layers = []
+        for i in range(num_shared_hidden):
+            if use_dropout:
+                shared_layers.append(keras.layers.Dropout(0.2 if i == 0 else 0.5))
+            shared_layers.append(keras.layers.Dense(hidden_dim, activation="relu"))
+            
+        self.shared = keras.Sequential(shared_layers, name='shared')
+
+        self.multitask_heads = []
+        for name in nutr_names:
+            head_layers = [
+                keras.layers.Dense(hidden_dim, activation="relu"),
+                keras.layers.Dense(1, activation="relu")
+            ]
+
+            if use_dropout:
+                head_layers.insert(0, keras.layers.Dropout(0.5))
+                head_layers.insert(2, keras.layers.Dropout(0.5))
+
+            self.multitask_heads.append(keras.Sequential(head_layers, name=name))
+
+    def call(self, inputs, training=False):
+        x = self.cnn(inputs, training=training)
+        x = self.shared(x, training=training)
+        return {head.name: head(x, training=training) for head in self.multitask_heads}

pipeline.py

@@ -0,0 +1,32 @@
+from typing import Dict, List, Any
+from PIL import Image
+from fcnutr import FCNutr
+
+import os
+import json
+import numpy as np
+import tensorflow as tf
+
+
+class PreTrainedPipeline():
+    def __init__(self, path=""):
+        crop_size = (224, 224)
+        self.nutr_names = ('energy', 'fat', 'protein', 'carbs')
+        self.model = FCNutr(self.nutr_names, crop_size, 4096, 3, False)
+        self.model.compile()
+        self.model(tf.zeros((1, crop_size[0], crop_size[1], 3)))
+        self.model.load_weights(os.path.join(path, "fcnutr.h5"))
+
+    def __call__(self, inputs: "Image.Image") -> List[Dict[str, Any]]:
+        image = tf.keras.preprocessing.image.img_to_array(inputs)
+        height = tf.shape(image)[0]
+        width = tf.shape(image)[1]
+        if width > height:
+            image = tf.image.resize(image, (self.img_size, int(float(self.img_size * width) / float(height))))
+        else:
+            image = tf.image.resize(image, (int(float(self.img_size * height) / float(width)), self.img_size))
+
+        image = tf.keras.applications.inception_v3.preprocess_input(image)
+        image = tf.keras.layers.CenterCrop(*self.crop_size)(image)
+        prediction = self.model(image[tf.newaxis, :])
+        return {name: float(prediction[name].numpy()[0, 0]) for name in self.nutr_names}