466 update context to include performance month and performance data and staff number (#469)

* staff_number, performance_month, performance_df, subject_graph moved to context

Author: faridseifi

scaffold/api.py

@@ -1,12 +1,10 @@
-import pandas as pd
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import RedirectResponse
 
 from scaffold import context
 from scaffold.pipeline import pipeline
 from scaffold.startup import startup
 from scaffold.utils.settings import settings
-from scaffold.utils.utils import get_performance_month
 
 app = FastAPI()
 
@@ -34,15 +32,10 @@ async def template():
 async def createprecisionfeedback(info: Request):
     req_info = await info.json()
 
-    performance_month = get_performance_month(req_info)
-    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, performance_df.at[0, "staff_number"], performance_month
-        )
+        context.from_req(req_info)
+
+        full_message = pipeline()
         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,6 +1,7 @@
 import pandas as pd
 from rdflib import RDF, BNode, Graph
 
+from scaffold import context
 from scaffold.bitstomach.signals import SIGNALS
 from scaffold.utils.namespace import PSDO, SLOWMO
 
@@ -31,18 +32,14 @@ def extract_signals(perf_df: pd.DataFrame) -> Graph:
     return g
 
 
-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"])
+def prepare():
+    performance_df = context.performance_df
 
     performance_df["goal_comparator_content"] = performance_df["MPOG_goal"]
 
-    performance_df = performance_df[performance_df["month"] <= performance_month]
+    performance_df = performance_df[
+        performance_df["month"] <= context.performance_month
+    ]
 
     performance_df["valid"] = performance_df["denominator"] >= 10
 
@@ -53,7 +50,10 @@ def prepare(performance_df, staff_number, performance_month):
     performance_df.attrs["measures"] = performance_df["measure"].unique()
 
     performance_df.attrs["valid_measures"] = performance_df[
-        ((performance_df["month"] == performance_month) & performance_df["valid"])
+        (
+            (performance_df["month"] == context.performance_month)
+            & performance_df["valid"]
+        )
     ]["measure"]
 
-    return performance_df, performance_month
+    return performance_df

scaffold/candidate_pudding/candidate_pudding.py

@@ -3,6 +3,7 @@
 from rdflib import RDF, BNode, Graph, Literal, URIRef
 from rdflib.resource import Resource
 
+from scaffold import context
 from scaffold.bitstomach.signals import Signal
 from scaffold.utils.namespace import CPO, IAO, PSDO, RO, SCHEMA, SLOWMO
 
@@ -110,16 +111,20 @@ def add_convenience_properties(candidate: Resource):
     return candidate
 
 
-def create_candidates(graph: Graph):
+def create_candidates():
     # measures = graph[: RDF.type : PSDO.performance_measure_content]
     # How do we get the measures for all MI?
     measures: set[BNode] = set(
-        graph.objects(None, PSDO.motivating_information / SLOWMO.RegardingMeasure)
+        context.subject_graph.objects(
+            None, PSDO.motivating_information / SLOWMO.RegardingMeasure
+        )
     )
     for measure in measures:
-        measure_resource = graph.resource(measure)
-        for template in graph[: RDF.type : PERFORMANCE_SUMMARY_DISPLAY_TEMPLATE]:
-            template_resource = graph.resource(template)
+        measure_resource = context.subject_graph.resource(measure)
+        for template in context.subject_graph[
+            : RDF.type : PERFORMANCE_SUMMARY_DISPLAY_TEMPLATE
+        ]:
+            template_resource = context.subject_graph.resource(template)
             candidate = create_candidate(measure_resource, template_resource)
             if not candidate:
                 continue

scaffold/cli.py

@@ -4,21 +4,18 @@
 from typing import Annotated
 
 import orjson
-import pandas as pd
 import typer
 from fastapi import HTTPException
 from loguru import logger
 
-from scaffold import context
+from scaffold import context, startup
 from scaffold.pipeline import pipeline
-from scaffold.startup import startup
 from scaffold.utils.utils import (
     add_candidates,
     add_response,
     analyse_candidates,
     analyse_responses,
     extract_number,
-    get_performance_month,
 )
 
 cli = typer.Typer(no_args_is_help=True)
@@ -41,7 +38,7 @@ def batch(
         ),
     ] = False,
 ) -> None:
-    startup()
+    startup.startup()
 
     if file_path.is_file() and file_path.suffix == ".json":
         input_files = [file_path]
@@ -60,18 +57,10 @@ def batch(
         try:
             input_data = orjson.loads(input_file.read_bytes())
 
-            performance_month = get_performance_month(input_data)
-            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,
-                    performance_df.at[0, "staff_number"],
-                    performance_month,
-                )
+                context.from_req(input_data)
+
+                full_message = pipeline()
                 full_message["message_instance_id"] = input_data["message_instance_id"]
                 full_message["performance_data"] = input_data["Performance_data"]
             except HTTPException as e:
@@ -82,9 +71,8 @@ def batch(
                 directory = input_file.parent / "messages"
                 os.makedirs(directory, exist_ok=True)
 
-                performance_month = input_data.get("performance_month", "unknown_month")
                 new_filename = (
-                    f"{input_file.stem} - message for {performance_month}.json"
+                    f"{input_file.stem} - message for {context.performance_month}.json"
                 )
                 output_path = directory / new_filename
 
@@ -103,7 +91,7 @@ def batch(
 
         add_response(full_message)
         if not stats_only:
-            add_candidates(full_message, input_data["performance_month"])
+            add_candidates(full_message)
 
     logger.info(f"Total files scanned: {len(input_files[:max_files])}")
     logger.info(f"Successful: {success_count}, Failed: {failure_count}")
@@ -132,43 +120,48 @@ def batch_csv(
         ),
     ] = False,
 ):
-    startup()
+    startup.startup(performance_data_path)
 
-    performance_data = pd.read_csv(performance_data_path, parse_dates=["month"])
     success_count = 0
     failure_count = 0
-    for provider_id in (
-        performance_data["staff_number"].drop_duplicates().head(max_files)
+    for staff_number in (
+        startup.performance_data["staff_number"].drop_duplicates().head(max_files)
     ):
         try:
-            context.create({}, provider_id)
-            result = pipeline(performance_data, provider_id, performance_month)
+            context.from_global(staff_number, performance_month)
+            try:
+                full_message = pipeline()
+                # full_message["message_instance_id"] = input_data["message_instance_id"]
+                full_message["performance_data"] = performance_month
+            except Exception as e:
+                # e.detail["message_instance_id"] = input_data["message_instance_id"]
+                raise e
             if not stats_only:
                 directory = performance_data_path.parent / "messages"
                 os.makedirs(directory, exist_ok=True)
 
                 new_filename = (
-                    f"Provider_{provider_id} - message for {performance_month}.json"
+                    f"Provider_{staff_number} - message for {performance_month}.json"
                 )
                 output_path = directory / new_filename
 
                 output_path.write_bytes(
-                    orjson.dumps(result, option=orjson.OPT_INDENT_2)
+                    orjson.dumps(full_message, option=orjson.OPT_INDENT_2)
                 )
                 logger.info(f"Message created at {output_path}")
             else:
-                logger.info(f"✔ Would process: Provider_{provider_id}")
+                logger.info(f"✔ Would process: Provider_{staff_number}")
 
             success_count += 1
 
         except Exception as e:
-            logger.error(f"✘ Failed to process Provider_{provider_id}: {e}")
+            logger.error(f"✘ Failed to process Provider_{staff_number}: {e}")
             failure_count += 1
-            result = e.detail
+            full_message = e.detail
 
-        add_response(result)
+        add_response(full_message)
         if not stats_only:
-            add_candidates(result, performance_month)
+            add_candidates(full_message)
 
     logger.info(f"Successful: {success_count}, Failed: {failure_count}")
     analyse_responses()

scaffold/context.py

@@ -1,35 +1,88 @@
+import pandas as pd
+from rdflib import Graph
+
 from scaffold import startup
+from scaffold.utils.utils import get_performance_month
 
 preferences_dict = {}
 history_dict = {}
+staff_number = 0
+performance_month = ""
+performance_df: pd.DataFrame
+subject_graph = Graph()
 
 
-def create(req_info, staff_number):
-    global preferences_dict, history_dict
+def from_req(req_info):
+    global \
+        preferences_dict, \
+        history_dict, \
+        staff_number, \
+        performance_month, \
+        performance_df, \
+        subject_graph
+
+    try:
+        performance_df = pd.DataFrame(
+            req_info["Performance_data"][1:], columns=req_info["Performance_data"][0]
+        )
+    except Exception:
+        pass
+
+    performance_month = get_performance_month(req_info, performance_df["month"].max())
+
+    staff_number = int(performance_df.at[0, "staff_number"])
 
-    history_dict = {}
     preferences_dict = {}
+    try:
+        preferences_dict = set_preferences(req_info.get("Preferences", {}))
+    except Exception:
+        return set_preferences({})
+
+    history_dict = {}
+    try:
+        history_dict = req_info.get("History", {})
+    except Exception:
+        pass
+
+    subject_graph = Graph()
+    subject_graph += startup.base_graph
 
+
+def from_global(staff_num, perf_month):
+    global \
+        preferences_dict, \
+        history_dict, \
+        staff_number, \
+        performance_month, \
+        performance_df, \
+        subject_graph
+
+    staff_number = int(staff_num)
+    history_dict = {}
+    preferences_dict = {}
+    performance_month = perf_month
     try:
-        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)
+        p = startup.preferences.loc[staff_number, "preferences"]
+        preferences_dict = set_preferences(p)
     except Exception:
         return set_preferences({})
 
     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()
+        staff_data = startup.history[startup.history["staff_number"] == staff_number]
+        history_dict = staff_data.set_index("month")["history"].to_dict()
+    except Exception:
+        pass
+
+    try:
+        performance_df = startup.performance_data[
+            startup.performance_data["staff_number"] == staff_number
+        ].reset_index(drop=True)
     except Exception:
         pass
 
+    subject_graph = Graph()
+    subject_graph += startup.base_graph
+
 
 def get_preferences():
     global preferences_dict