skeleton

app.py

@@ -20,13 +20,13 @@ def compute_scores(user_gen_designs: pd.DataFrame) -> ModelScoringResult:
 
 
 def process_generated_designs(github_link: str, file: PathLike[str]):
-    uuid4 = uuid.uuid4()
     validate_github_link(github_link)
     with open(file, 'r') as user_file:
         user_gen_designs = pd.read_csv(user_file)
     validate_user_designs(user_gen_designs)
-    REPOSITORY_INSTANCE.add_row(compute_scores(user_gen_designs))
-    return "File uploaded successfully, uuid {uuid4}"
+    scores = compute_scores(user_gen_designs)
+    REPOSITORY_INSTANCE.add_row(scores)
+    return f"File uploaded successfully, uuid {scores.uuid}"
 
 
 with gr.Blocks() as gradio_app:

config.py

@@ -4,18 +4,20 @@ import attrs
 
 
 @attrs.define(frozen=True)
-class AppConfig:
+class AppConfigModel:
     production: bool
 
 def get_env_or_warn(key: str):
     result = os.getenv(key)
-
+    if (result is None) or (result.strip() == ""):
+        print(f"Warning: could not find environment variable {key}")
+        return ""
     return result
 
 
-def get_app_config() -> AppConfig:
-    return AppConfig(
-        production=os.getenv("DEPLOYMENT_ENVIRONMENT").upper() == "PRODUCTION"
+def get_app_config() -> AppConfigModel:
+    return AppConfigModel(
+        production=get_env_or_warn("DEPLOYMENT_ENVIRONMENT").upper() == "PRODUCTION"
     )
 
 

data_repository.py

@@ -6,7 +6,7 @@ from typing import List
 import attrs
 import pandas as pd
 
-from config import AppConfig
+from config import APP_CONFIG
 
 
 @attrs.define
@@ -18,49 +18,58 @@ class ModelScoringResult:
 
 
 ORDERED_COLUMNS = [
-    "model_id",
-    "model_score",
+    "uuid",
+    "score",
     "submission_time",
+    "scoring_time",
 ]
 
+class PandasModelScoresRepository(metaclass=abc.ABCMeta):
 
+    def get_data_to_display(self):
+        return pd.DataFrame(self.read_curr_state(), columns=ORDERED_COLUMNS)
 
-class ModelScoresRepository(metaclass=abc.ABCMeta):
+    def add_row(self, row: ModelScoringResult):
+        previous_state = self.read_curr_state()
+        result = pd.concat([previous_state, pd.DataFrame(attrs.asdict(row), index=range(1))])
+        self.save_to_disk(result)
 
-    @abc.abstractmethod
-    def get_data_to_display(self) -> pd.DataFrame:
-        pass
+    def save_current_state(self, rows: List[ModelScoringResult]):
+        self.save_to_disk(pd.DataFrame([attrs.asdict(r) for r in rows]))
 
     @abc.abstractmethod
-    def add_row(self, row: ModelScoringResult) -> None:
+    def save_to_disk(self, result: pd.DataFrame):
         pass
 
     @abc.abstractmethod
-    def save_current_state(self, rows: List[ModelScoringResult]) -> None:
+    def read_curr_state(self) -> pd.DataFrame:
         pass
 
 
-class LocalModelScoresRepository(ModelScoresRepository):
+class LocalPandasModelScoresRepository(PandasModelScoresRepository):
     def __init__(self, dummy_file_path=os.path.join(os.path.dirname(__file__), "dummy_data.txt")):
         self.dummy_file_path = dummy_file_path
         if not os.path.exists(self.dummy_file_path):
             with open(self.dummy_file_path, "w") as file:
                 file.write("uuid,score,submission_time,scoring_time")
 
-    def get_data_to_display(self):
-        return pd.DataFrame(pd.read_csv(self.dummy_file_path), columns=ORDERED_COLUMNS)
+    def read_curr_state(self) -> pd.DataFrame:
+        return pd.read_csv(self.dummy_file_path, index_col=None)
 
-    def add_row(self, row: ModelScoringResult):
-        previous_state = pd.read_csv(self.dummy_file_path)
-        result = pd.concat([previous_state, pd.DataFrame(row.as_dict())])
-        result.to_csv(self.dummy_file_path)
+    def save_to_disk(self, result: pd.DataFrame):
+        result.to_csv(self.dummy_file_path, index=False)
 
-    def save_current_state(self, rows: List[ModelScoringResult]):
-        pd.DataFrame([r.as_dict() for r in rows]).to_csv(self.dummy_file_path)
 
+class HuggingFaceDatasetModelScoresRepository(PandasModelScoresRepository):
+    def read_curr_state(self) -> pd.DataFrame:
+        pass
 
-class HuggingFaceDatasetModelScoresRepository(ModelScoresRepository):
-    pass
+    def save_to_disk(self, result: pd.DataFrame):
+        pass
 
 
-REPOSITORY_INSTANCE: ModelScoresRepository = HuggingFaceDatasetModelScoresRepository() if AppConfig.production else LocalModelScoresRepository()
+REPOSITORY_INSTANCE: PandasModelScoresRepository
+if APP_CONFIG.production:
+    REPOSITORY_INSTANCE = HuggingFaceDatasetModelScoresRepository()
+else:
+    REPOSITORY_INSTANCE = LocalPandasModelScoresRepository()