464 update data handling (#465)

* moved prepare call inside pipeline.
cleaned up attachment of performance month to performance data frame.
moved responsibility of cleaning up performance data for a specific provider to the prepare sending as input all performance data in a df with staff number and have it clean up the data frame as first step.

* simplified context update, get preferences and get history and moved history and preferences load to startup.

* context changed to have a create method which would create and store preferences and history from request or loaded files in context dictionaries. Get preferences and history methods are updated accordingly and context.create is called in api and batch modes.

Author: faridseifi

bulk-up/src/bulk_up/log_to_data.py

@@ -110,7 +110,9 @@ def add_signal_properties(row, output_message, input_message):
         columns=input_message["Performance_data"][0],
     )
 
-    performance_df = prepare(performance_month, performance_df)
+    performance_df, performance_month = prepare(
+        performance_df, performance_df.at[0, "staff_number"], performance_month
+    )
     performance_df = performance_df[
         performance_df["measure"] == output_message["selected_candidate"]["measure"]
     ].tail(12)

bulk-up/src/bulk_up/prepare_csv_inputs.py

@@ -25,43 +25,17 @@ def extract_number(filename):
             writer.writerows(input_data["Performance_data"][1:])
 
 
-fieldnames = [
-    "staff_number",
-    "Social gain",
-    "Social stayed better",
-    "Worsening",
-    "Improving",
-    "Social loss",
-    "Social stayed worse",
-    "Social better",
-    "Social worse",
-    "Social approach",
-    "Goal gain",
-    "Goal approach",
-    "Display_Format",
-]
 with open("preferences.csv", "w", newline="") as file:
-    writer = csv.DictWriter(file, fieldnames=fieldnames)
+    writer = csv.DictWriter(file, fieldnames=["staff_number", "preferences"])
     writer.writeheader()
     for index, input_file in enumerate(input_files):
         input_data = orjson.loads(input_file.read_bytes())
-        if input_data["Preferences"].get("Utilities", {}).get("Message_Format", {}):
-            preferences = {"staff_number": input_data["Performance_data"][1][0]}
-            preferences.update(
-                input_data["Preferences"].get("Utilities", {}).get("Message_Format", {})
-            )
-            preferences["Display_Format"] = next(
-                (
-                    k
-                    for k, v in input_data["Preferences"]
-                    .get("Utilities", {})
-                    .get("Display_Format", {})
-                    .items()
-                    if v == 1
-                ),
-                None,
-            )
-            writer.writerows([preferences])
+        preferences = {
+            "staff_number": input_data["Performance_data"][1][0],
+            "preferences": orjson.dumps(input_data["Preferences"]).decode(),
+        }
+
+        writer.writerows([preferences])
 
 all_keys = set(["staff_number", "month", "history"])
 for input_file in input_files:
@@ -75,6 +49,6 @@ def extract_number(filename):
             history = {
                 "staff_number": input_data["Performance_data"][1][0],
                 "month": key,
-                "history": value,
+                "history": orjson.dumps(value).decode(),
             }
             writer.writerows([history])

scaffold/api.py

@@ -3,7 +3,6 @@
 from fastapi.responses import RedirectResponse
 
 from scaffold import context
-from scaffold.bitstomach.bitstomach import prepare
 from scaffold.pipeline import pipeline
 from scaffold.startup import startup
 from scaffold.utils.settings import settings
@@ -34,17 +33,16 @@ async def template():
 @app.post("/createprecisionfeedback/")
 async def createprecisionfeedback(info: Request):
     req_info = await info.json()
-    context.update(req_info)
 
     performance_month = get_performance_month(req_info)
-    performance_df = prepare(
-        performance_month,
-        pd.DataFrame(
-            req_info["Performance_data"][1:], columns=req_info["Performance_data"][0]
-        ),
+    performance_df = pd.DataFrame(
+        req_info["Performance_data"][1:], columns=req_info["Performance_data"][0]
     )
+    context.create(req_info, performance_df.at[0, "staff_number"])
     try:
-        full_message = pipeline(performance_df)
+        full_message = pipeline(
+            performance_df, performance_df.at[0, "staff_number"], performance_month
+        )
         full_message["message_instance_id"] = req_info["message_instance_id"]
         full_message["performance_data"] = req_info["Performance_data"]
     except HTTPException as e:

scaffold/bitstomach/bitstomach.py

@@ -1,5 +1,5 @@
 import pandas as pd
-from rdflib import RDF, BNode, Graph, Literal
+from rdflib import RDF, BNode, Graph
 
 from scaffold.bitstomach.signals import SIGNALS
 from scaffold.utils.namespace import PSDO, SLOWMO
@@ -14,7 +14,6 @@ def extract_signals(perf_df: pd.DataFrame) -> Graph:
     g = Graph()
     r = g.resource(BNode("performance_content"))
     r.set(RDF.type, PSDO.performance_content)
-    r.set(SLOWMO.PerformanceMonth, Literal(perf_df.attrs["performance_month"]))
     if perf_df.empty:
         return g
 
@@ -32,19 +31,18 @@ def extract_signals(perf_df: pd.DataFrame) -> Graph:
     return g
 
 
-def prepare(performance_month, performance_df):
-    # we would have multiple staff performance data at this point so this like won't work right
+def prepare(performance_df, staff_number, performance_month):
+    performance_df = performance_df[
+        performance_df["staff_number"] == staff_number
+    ].reset_index(drop=True)
+
+    if not performance_month:
+        performance_month = performance_df["month"].max()
     performance_df.attrs["staff_number"] = int(performance_df.at[0, "staff_number"])
 
     performance_df["goal_comparator_content"] = performance_df["MPOG_goal"]
 
-    performance_df.attrs["performance_month"] = (
-        performance_month if performance_month else performance_df["month"].max()
-    )
-
-    performance_df = performance_df[
-        performance_df["month"] <= performance_df.attrs["performance_month"]
-    ]
+    performance_df = performance_df[performance_df["month"] <= performance_month]
 
     performance_df["valid"] = performance_df["denominator"] >= 10
 
@@ -55,10 +53,7 @@ def prepare(performance_month, performance_df):
     performance_df.attrs["measures"] = performance_df["measure"].unique()
 
     performance_df.attrs["valid_measures"] = performance_df[
-        (
-            (performance_df["month"] == performance_df.attrs["performance_month"])
-            & performance_df["valid"]
-        )
+        ((performance_df["month"] == performance_month) & performance_df["valid"])
     ]["measure"]
 
-    return performance_df
+    return performance_df, performance_month

scaffold/cli.py

@@ -10,7 +10,6 @@
 from loguru import logger
 
 from scaffold import context
-from scaffold.bitstomach.bitstomach import prepare
 from scaffold.pipeline import pipeline
 from scaffold.startup import startup
 from scaffold.utils.utils import (
@@ -61,18 +60,18 @@ def batch(
         try:
             input_data = orjson.loads(input_file.read_bytes())
 
-            context.update(input_data)
-
             performance_month = get_performance_month(input_data)
-            performance_df = prepare(
-                performance_month,
-                pd.DataFrame(
-                    input_data["Performance_data"][1:],
-                    columns=input_data["Performance_data"][0],
-                ),
+            performance_df = pd.DataFrame(
+                input_data["Performance_data"][1:],
+                columns=input_data["Performance_data"][0],
             )
+            context.create(input_data, performance_df.at[0, "staff_number"])
             try:
-                full_message = pipeline(performance_df)
+                full_message = pipeline(
+                    performance_df,
+                    performance_df.at[0, "staff_number"],
+                    performance_month,
+                )
                 full_message["message_instance_id"] = input_data["message_instance_id"]
                 full_message["performance_data"] = input_data["Performance_data"]
             except HTTPException as e:
@@ -134,7 +133,6 @@ def batch_csv(
     ] = False,
 ):
     startup()
-    context.init()
 
     performance_data = pd.read_csv(performance_data_path, parse_dates=["month"])
     success_count = 0
@@ -143,13 +141,8 @@ def batch_csv(
         performance_data["staff_number"].drop_duplicates().head(max_files)
     ):
         try:
-            performance_df = prepare(
-                performance_month,
-                performance_data[
-                    performance_data["staff_number"] == provider_id
-                ].reset_index(drop=True),
-            )
-            result = pipeline(performance_df)
+            context.create({}, provider_id)
+            result = pipeline(performance_data, provider_id, performance_month)
             if not stats_only:
                 directory = performance_data_path.parent / "messages"
                 os.makedirs(directory, exist_ok=True)

scaffold/context.py

@@ -1,141 +1,52 @@
-import json
+from scaffold import startup
 
-import numpy as np
-import pandas as pd
+preferences_dict = {}
+history_dict = {}
 
-from scaffold import startup
-from scaffold.utils.settings import settings
 
-preferences: pd.DataFrame = pd.DataFrame()
-history: pd.DataFrame = pd.DataFrame()
+def create(req_info, staff_number):
+    global preferences_dict, history_dict
 
+    history_dict = {}
+    preferences_dict = {}
 
-def init():
     try:
-        global preferences, history
-
-        if settings.preferences is not None:
-            preferences = pd.read_csv(settings.preferences)
-        if settings.history is not None:
-            history = pd.read_csv(settings.history, converters={"history": json.loads})
+        if req_info.get("Preferences", {}):
+            preferences_dict = set_preferences(req_info.get("Preferences", {}))
+        else:
+            p = startup.preferences.loc[staff_number, "preferences"]
+            preferences_dict = set_preferences(p)
+    except Exception:
+        return set_preferences({})
 
-    except Exception as e:
-        print("context init aborted, see traceback:")
-        raise e
+    try:
+        if req_info.get("History", {}):
+            history_dict = req_info.get("History", {})
+        else:
+            staff_data = startup.history[
+                startup.history["staff_number"] == staff_number
+            ]
+            history_dict = staff_data.set_index("month")["history"].to_dict()
+    except Exception:
+        pass
 
 
-def update(req_info):
-    try:
-        global preferences, history
-        staff_number = req_info["Performance_data"][1][0]
+def get_preferences():
+    global preferences_dict
+    return preferences_dict
 
-        preferences_dict = req_info.get("Preferences", {}).get("Utilities", {})
-        if preferences_dict:
-            new_row = {"staff_number": staff_number} | preferences_dict[
-                "Message_Format"
-            ]
-            display_format = next(
-                (
-                    k
-                    for k, v in preferences_dict.get("Display_Format", {}).items()
-                    if v == 1
-                ),
-                "None",
-            )
-            new_row["Display_Format"] = display_format
-
-            if preferences.empty:
-                preferences = pd.DataFrame([new_row])
-            elif not (preferences["staff_number"] == staff_number).any():
-                preferences = pd.concat(
-                    [preferences, pd.DataFrame([new_row])], ignore_index=True
-                )
-
-        history_dict: dict = req_info.get("History", {})
-
-        if history.empty:
-            history = pd.DataFrame(columns=["staff_number", "month", "history"])
-
-        for key, value in history_dict.items():
-            if (
-                history.empty
-                or history[
-                    (history["staff_number"] == staff_number)
-                    & (history["month"] == key)
-                ].empty
-            ):
-                new_row = pd.DataFrame(
-                    [{"staff_number": staff_number, "month": key, "history": value}]
-                )
-                history = pd.concat([history, pd.DataFrame(new_row)], ignore_index=True)
-    except Exception as e:
-        print("context update aborted, see traceback:")
-        raise e
-
-
-def get_preferences(staff_number):
-    global preferences
-    if preferences.empty:
-        return set_preferences({})
 
-    preferences_row = preferences[preferences["staff_number"] == staff_number]
-    provider_preferences = {}
-    if not preferences_row.empty:
-        provider_preferences = {
-            "Preferences": {"Utilities": {"Message_Format": {}, "Display_Format": {}}}
-        }
-
-        # We'll just use the first row of the CSV
-        row = preferences_row.iloc[0]
-
-        for key in preferences_row.columns:
-            value = row[key]
-            if key == "staff_number":
-                continue  # skip or store if you need it
-            elif key == "Display_Format":
-                # Example: "Bar chart, Line chart"
-                provider_preferences["Preferences"]["Utilities"]["Display_Format"] = {
-                    "Bar chart": 0,
-                    "Line chart": 0,
-                    "Text-only": 0,
-                    "System-generated": "0",
-                }
-                provider_preferences["Preferences"]["Utilities"]["Display_Format"][
-                    value
-                ] = 1
-            else:
-                if isinstance(value, (np.float64, np.int64)):
-                    value = value.item()
-                provider_preferences["Preferences"]["Utilities"]["Message_Format"][
-                    key
-                ] = value
-    return set_preferences(provider_preferences)
-
-
-def get_history(staff_number):
-    global history
-
-    if history.empty:
-        return {}
-    history_rows = history[history["staff_number"] == staff_number]
-    if history_rows.empty:
-        return {}
-
-    provider_history = {}
-    for index, row in history_rows.iterrows():
-        history_item = row["history"]  # ast.literal_eval(row["history"])
-        month = row["month"]
-        provider_history[month] = history_item
-
-    return provider_history
+def get_history():
+    global history_dict
+    return history_dict
 
 
 def set_preferences(req_info):
-    preferences_utilities = req_info.get("Preferences", {}).get("Utilities", {})
+    preferences_utilities = req_info.get("Utilities", {})
     input_preferences: dict = preferences_utilities.get("Message_Format", {})
     individual_preferences: dict = {}
     for key in input_preferences:
-        individual_preferences[key] = float(input_preferences[key])
+        individual_preferences[key.lower()] = float(input_preferences[key])
 
     preferences: dict = startup.default_preferences.copy()
     preferences.update(individual_preferences)