Max Ratings API: Updates from eCFR for digestive systems. (#3049)

domain-ee/ee-max-cfi-app/scripts/.gitignore

@@ -0,0 +1,6 @@
+# Ignore everything in the directory
+*
+
+# But don't ignore data_consolidator
+!dc_lookup_table_updater.py
+!README.md

domain-ee/ee-max-cfi-app/scripts/README.md

@@ -0,0 +1,23 @@
+# Diagnostic Code Lookup Table Updater
+
+## Overview
+
+The `dc_lookup_table_updater.py` script is used to update the Diagnostic Code Lookup Table in the database. The script reads a CSV file containing the new data
+and updates the database with the new data. The script also logs the changes made to the database.
+
+# Using the dc_lookup_table_updater.py script
+
+1. Prepare the new data CSV file and add it to this directory. The CSV file must have the following columns:
+    * Diagnostic Code **(Required)**
+    * Rated Issue Name
+    * Max Rating
+    * Body System
+    * Category
+    * Subcategory
+    * CFR Reference
+      Preparation includes removing any extra punctuation in the `Rated Issue Name` column.
+
+2. Run the script with the following command from the directory `domain-ee/ee-max-cfi-app/scripts`
+    ```
+    python dc_lookup_table_updater.py <new_data_csv_file>.csv
+    ```

domain-ee/ee-max-cfi-app/scripts/dc_lookup_table_updater.py

@@ -0,0 +1,77 @@
+import csv
+import sys
+from typing import Any
+
+TABLE_VERSION_PY = '../src/python_src/util/data/table_version.py'
+DATA_FILE = '../src/python_src/util/data/Diagnostic Code Lookup Table.csv'
+
+
+def consolidate_data_files(new_data_file: str) -> dict[int, tuple[Any, Any, Any, Any, Any, Any]] | None:
+    original_data = import_file(DATA_FILE)
+    data_map: dict[int, tuple[Any, int, Any, Any, Any, Any]] = original_data.copy()
+
+    imported_data = import_file(new_data_file)
+    updates = imported_data & data_map.keys()
+    adds = updates ^ imported_data.keys()
+    msg = f'Loading {len(imported_data)} ratings from "{new_data_file}" | Adds {len(adds)} | Potential Updates {len(updates)}'
+    print(msg)
+
+    data_map.update(imported_data)
+
+    new_data = dict(sorted(data_map.items()))
+
+    if new_data == original_data:
+        print('No changes detected')
+        return None
+
+    return new_data
+
+
+def import_file(filename: str) -> dict[int, tuple[Any, Any, Any, Any, Any, Any]]:
+    diagnostic_code_to_data: dict[int, tuple[Any, Any, Any, Any, Any, Any]] = {}
+    with open(filename, 'r') as file:
+        csv_reader = csv.reader(file)
+        for index, csv_line in enumerate(csv_reader):
+            if index == 0:
+                continue
+
+            if len(csv_line) != 7:
+                raise ValueError(f'Invalid CSV line at index {index}')
+
+            diagnostic_code_str, rated_issue_name, max_rating_str, body_system, category, subcategory, cfr_ref = csv_line
+
+            try:
+                diagnostic_code = int(diagnostic_code_str)
+            except ValueError:
+                raise ValueError(f'Invalid diagnostic code at index {index}: \n{csv_line}')
+
+            diagnostic_code_to_data[diagnostic_code] = (rated_issue_name, max_rating_str, body_system, category, subcategory, cfr_ref)
+
+    return diagnostic_code_to_data
+
+
+def export_data(data: dict[int, tuple[Any, int, Any, Any, Any, Any]]):
+    with open(DATA_FILE, 'w') as file:
+        csv_writer = csv.writer(file)
+        csv_writer.writerow(['Diagnostic Code', 'Rated Issue Name', 'Max Rating', 'Body System', 'Category', 'Subcategory', 'CFR Reference'])
+        for diagnostic_code, data_tuple in data.items():
+            csv_writer.writerow([diagnostic_code, *data_tuple])
+
+
+def increment_table_version():
+    with open(TABLE_VERSION_PY, 'r') as file:
+        version = int(file.readline().strip().split(' = ').pop())
+    with open(TABLE_VERSION_PY, 'w') as file:
+        file.write(f'TABLE_VERSION = {version + 1}\n')
+        file.write('# auto-generated by dc_lookup_table_updater.py\n')
+
+
+if __name__ == '__main__':
+    new_data_file = sys.argv[1]
+    print(new_data_file)
+    data = consolidate_data_files(new_data_file)
+    if data:
+        export_data(data)
+        print('Data exported')
+        increment_table_version()
+        print('Table version incremented')

domain-ee/ee-max-cfi-app/src/python_src/api.py

@@ -1,13 +1,11 @@
-import logging
-import sys
-
 import uvicorn
 from fastapi import FastAPI, HTTPException
 from pydantic_models import (
     MaxRatingsForClaimForIncreaseRequest,
     MaxRatingsForClaimForIncreaseResponse,
     Rating,
 )
+from util.logger import logger
 from util.lookup_table import MAX_RATINGS_BY_CODE, get_max_rating
 from util.sanitizer import sanitize
 
@@ -25,13 +23,6 @@
     ],
 )
 
-logging.basicConfig(
-    format='[%(asctime)s] %(levelname)-8s %(message)s',
-    level=logging.INFO,
-    datefmt='%Y-%m-%d %H:%M:%S',
-    stream=sys.stdout,
-)
-
 
 @app.get('/health')
 def get_health_status() -> dict[str, str]:
@@ -55,7 +46,7 @@ def get_max_ratings(
 
     response = MaxRatingsForClaimForIncreaseResponse(ratings=ratings)
 
-    logging.info(f'event=getMaxRating ratings={ratings}')
+    logger.info(f'event=getMaxRating response={response.model_dump_json()}')
     return response
 
 

domain-ee/ee-max-cfi-app/src/python_src/pydantic_models.py

@@ -1,9 +1,9 @@
-from pydantic import BaseModel, StrictInt, Field
+from pydantic import BaseModel, Field, StrictInt
 from typing_extensions import Annotated
 
 
 class MaxRatingsForClaimForIncreaseRequest(BaseModel):
-    diagnostic_codes: Annotated[list[StrictInt], Field(max_items=1000)]
+    diagnostic_codes: Annotated[list[StrictInt], Field(max_length=1000)]
 
 
 class Rating(BaseModel):