update

Author: TJaniF

.gitignore

@@ -7,4 +7,5 @@ astro
 z_experiments
 include/duckdb_database
 include/pretained_models/
-db_script.py
+db_script.py
+include/minio

dags/deploy_best_model.py

@@ -5,35 +5,14 @@
 """
 
 from airflow import Dataset as AirflowDataset
-from airflow.decorators import dag, task_group, task
-from astro import sql as aql
-from astro.sql import get_value_list
-from astro.files import get_file_list
-from astro.sql.table import Table
+from airflow.decorators import dag, task
 from airflow.operators.empty import EmptyOperator
-from airflow.operators.bash import BashOperator
-from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.models import Variable
-
-from collections import Counter
-import pandas as pd
 from pendulum import datetime
-import os
 import logging
-import requests
-import numpy as np
-from PIL import Image
 import duckdb
-import json
-import pickle
-import shutil
-import torch
 from airflow.sensors.base import PokeReturnValue
 
-from include.custom_operators.hugging_face import (
-    TestHuggingFaceImageClassifierOperator,
-    transform_function,
-)
 
 task_logger = logging.getLogger("airflow.task")
 
@@ -79,13 +58,22 @@ def pick_best_model_from_db(db_path):
                 FROM model_results
                 WHERE test_set_num = (SELECT MAX(test_set_num) FROM model_results)
                 ORDER BY test_av_loss ASC
-                LIMIT 1;"""
-        ).fetchall()[0][0]
+                LIMIT 1;""" # want higher false negative - be more sensitive recall!!. maybe optimized for f score (long discussion on medium), want to use precision or recall
+        ).fetchall()[0][0] #ROC area
         con.close()
 
         return best_model_latest_test_set
 
-    ensure_baseline_ran() >> pick_best_model_from_db(db_path=DUCKDB_PATH)
+    @task
+    def deploy_model(model):
+        pass
+
+    (
+        start
+        >> ensure_baseline_ran()
+        >> deploy_model(pick_best_model_from_db(db_path=DUCKDB_PATH))
+        >> end
+    )
 
 
 deploy_best_model()

dags/in_new_test_data.py

@@ -28,7 +28,8 @@
 
 @dag(
     start_date=datetime(2023, 1, 1),
-    schedule=None,
+    schedule="@continuous",
+    max_active_runs=1,
     catchup=False,
 )
 def in_new_test_data():

dags/in_new_train_data.py

@@ -28,7 +28,8 @@
 
 @dag(
     start_date=datetime(2023, 1, 1),
-    schedule=None,
+    schedule="@continuous",
+    max_active_runs=1,
     catchup=False,
 )
 def in_new_train_data():

dags/preprocess_test_data.py

@@ -5,25 +5,15 @@
 """
 
 from airflow import Dataset
-from airflow.decorators import dag, task_group, task
-from astro import sql as aql
-from astro.sql import get_value_list
+from airflow.decorators import dag, task
 from astro.files import get_file_list
-from astro.sql.table import Table
 from airflow.operators.empty import EmptyOperator
 from airflow.operators.bash import BashOperator
 from airflow.models import Variable
 
-import pandas as pd
 from pendulum import datetime
-import os
 import logging
-import requests
-import numpy as np
-from PIL import Image
 import duckdb
-import json
-import pickle
 
 task_logger = logging.getLogger("airflow.task")
 

dags/preprocess_train_data.py

@@ -5,7 +5,7 @@
 """
 
 from airflow import Dataset
-from airflow.decorators import dag, task_group, task
+from airflow.decorators import dag, task
 from astro import sql as aql
 from astro.sql import get_value_list
 from astro.files import get_file_list

dags/test_fine_tuned_model.py

@@ -20,6 +20,7 @@
 import pickle
 import shutil
 import torch
+from airflow.models import Variable
 
 from include.custom_operators.hugging_face import (
     TestHuggingFaceImageClassifierOperator,
@@ -161,15 +162,16 @@ def write_model_results_to_duckdb(db_path, table_name, **context):
         model_name = context["ti"].xcom_pull(task_ids="test_classifier")["model_name"]
 
         con = duckdb.connect(db_path)
+        test_set_num = Variable.get("test_set_num")
 
         con.execute(
             f"""CREATE TABLE IF NOT EXISTS {table_name} 
-            (model_name TEXT PRIMARY KEY, timestamp DATETIME, test_av_loss FLOAT, test_accuracy FLOAT)"""
+            (model_name TEXT PRIMARY KEY, timestamp DATETIME, test_av_loss FLOAT, test_accuracy FLOAT, test_set_num INT)"""
         )
 
         con.execute(
-            f"INSERT OR REPLACE INTO {table_name} (model_name, timestamp, test_av_loss, test_accuracy) VALUES (?, ?, ?, ?) ",
-            (model_name, timestamp, test_av_loss, test_accuracy),
+            f"INSERT OR REPLACE INTO {table_name} (model_name, timestamp, test_av_loss, test_accuracy, test_set_num) VALUES (?, ?, ?, ?, ?) ",
+            (model_name, timestamp, test_av_loss, test_accuracy, test_set_num),
         )
 
         con.close()

dags/train_model.py

@@ -102,8 +102,8 @@ def load_training_images(keys):
 
     train_classifier = TrainHuggingFaceImageClassifierOperator(
         task_id="train_classifier",
-        model_name="microsoft/resnet-50",
-        criterion=torch.nn.CrossEntropyLoss(),
+        model_name="microsoft/resnet-50", # find newer one?
+        criterion=torch.nn.CrossEntropyLoss(),  # binary entropy loss!
         optimizer=torch.optim.Adam,
         local_images_filepaths=local_images_filepaths,
         labels=get_labels_from_duckdb.map(lambda x: x[0]),

include/custom_operators/hugging_face.py

@@ -100,6 +100,8 @@ class TrainHuggingFaceImageClassifierOperator(BaseOperator):
 
     """
 
+    ui_color = "#91ed9d"
+
     template_fields = (
         "model_name",
         "criterion",
@@ -131,7 +133,7 @@ def __init__(
         super().__init__(*args, **kwargs)
         self.model_name = model_name
         self.criterion = criterion
-        self.optimizer = optimizer
+        self.optimizer = optimizer #change optimizer
         self.local_images_filepaths = local_images_filepaths
         self.labels = labels
         self.num_classes = num_classes
@@ -154,16 +156,17 @@ def execute(self, context):
             num_workers=0,
         )
 
-        model = ResNetForImageClassification.from_pretrained(self.model_name)
+        # figure out how fine tuning happens inside hugging face
+        model = ResNetForImageClassification.from_pretrained(self.model_name) # does this do something optimized for fine tune or not?
         model.classifier[-1] = torch.nn.Linear(
             model.classifier[-1].in_features, self.num_classes
         )
 
         device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
         model = model.to(device)
-        optimizer = self.optimizer(model.parameters(), lr=1e-4)
+        optimizer = self.optimizer(model.parameters(), lr=1e-4) ### lr is a hyperparameter learning rate to be adjusted. add as a parameter
 
-        for epoch in range(self.num_epochs):
+        for epoch in range(self.num_epochs): #num of epochs
             model.train()
             running_loss = 0.0
             running_corrects = 0
@@ -206,6 +209,8 @@ class TestHuggingFaceImageClassifierOperator(BaseOperator):
 
     """
 
+    ui_color = "#ebab34"
+
     template_fields = (
         "model_name",
         "criterion",