Calculate index value

.github/workflows/_pixi-lock-check.yml

@@ -17,6 +17,7 @@ jobs:
           - plugins/flows/base
           - plugins/flows/data_management
           - plugins/flows/data_transformation
+          - plugins/flows/eq5d5l_index_calculation
           - plugins/flows/hades
           - plugins/flows/i2b2
           - plugins/flows/loyalty_score

.github/workflows/docker-build-push.yaml

@@ -339,6 +339,7 @@ jobs:
           - PKGPATH: ./plugins/flows/data_management/
           - PKGPATH: ./plugins/flows/hades/
           - PKGPATH: ./plugins/flows/loyalty_score/
+          - PKGPATH: ./plugins/flows/eq5d5l_index_calculation/
           - PKGPATH: ./plugins/flows/data_transformation/
           - PKGPATH: ./plugins/ui
             NPM: bun

docker-compose-local.yml

@@ -83,10 +83,12 @@ services:
       # services/trex/core) is bundled into the image's eszip by Dockerfile.v2.
       # - ./services/trex/deno.json:/usr/src/deno.json
       # - ./plugins/flows/base/package.json:/usr/src/plugins/d2e-flows/package.json # For local flows development only
+      - ./plugins/flows/eq5d5l_index_calculation/package.json:/usr/src/plugins/d2e-flows/package.json
     env_file:
       - path: .env.claw
         required: false
     environment:
+      PLUGINS_SEED_UPDATE: true
       TREX_DX_ENABLED: "true"
       # Enable trex's native email/password IDP for local dev (off by default in
       # the engine). Production d2e authenticates via Logto and leaves it unset.

plugins/flows/_shared_flow_utils/dao/sqlalchemydao.py

@@ -206,6 +206,21 @@ def get_next_record_id(self, schema: str, table: str, id_column: int) -> int:
         else:
             return int(last_record_id) + 1
 
+    def select_rows_where_in(
+        self, schema: str, table: str, columns: list[str], where_column: str, where_values: list
+    ) -> list[dict]:
+        """
+        Select `columns` from `table` where `where_column` is in `where_values`.
+        Returns one dict per row, keyed by the requested column names.
+        """
+        with self.engine.connect() as connection:
+            metadata_obj = sql.MetaData(schema=schema)
+            table_obj = sql.Table(table, metadata_obj, autoload_with=connection)
+            select_cols = [table_obj.c[col] for col in columns]
+            stmt = sql.select(*select_cols).where(table_obj.c[where_column].in_(where_values))
+            result = connection.execute(stmt).mappings().all()
+        return [dict(row) for row in result]
+
     # --- Update methods ---
 
     def update_cdm_version(self, schema: str, cdm_version: str):
@@ -232,6 +247,48 @@ def insert_values_into_table(
             res = connection.execute(table_obj.insert(), column_value_mapping)
             connection.commit()
 
+    def delete_and_insert_rows(
+        self,
+        schema: str,
+        table: str,
+        delete_column: str,
+        delete_value,
+        insert_rows: list[dict],
+        id_column: str = None,
+    ) -> list[dict]:
+        """
+        Atomically (single transaction): delete every row where delete_column ==
+        delete_value, then insert insert_rows. If id_column is given, each inserted
+        row is assigned a sequential id continuing from the table's current max
+        (computed after the delete, in the same transaction) - the returned rows
+        carry that assigned id_column value. Use this instead of a separate
+        delete_records()/insert_values_into_table() pair when the delete and
+        insert must not be observable as two separate commits (e.g. an
+        overwrite-on-rerun that must never leave the table with the old rows
+        deleted and nothing inserted in their place).
+        """
+        with self.engine.begin() as connection:
+            metadata_obj = sql.MetaData(schema=schema)
+            table_obj = sql.Table(table, metadata_obj, autoload_with=connection)
+
+            connection.execute(
+                table_obj.delete().where(table_obj.c[delete_column] == delete_value)
+            )
+
+            if id_column:
+                last_id = connection.execute(
+                    sql.select(sql.func.max(table_obj.c[id_column]))
+                ).scalar()
+                next_id = (int(last_id) + 1) if last_id is not None else 1
+                insert_rows = [
+                    {**row, id_column: next_id + i} for i, row in enumerate(insert_rows)
+                ]
+
+            if insert_rows:
+                connection.execute(table_obj.insert(), insert_rows)
+
+        return insert_rows
+
     def update_data_ingestion_date(self, schema: str):
         with self.engine.connect() as connection:
             metadata_obj = sql.MetaData(schema=schema)
@@ -310,6 +367,10 @@ def truncate_table(self, schema: str, table: str):
     def return_affected_rowcounts(result) -> int:
         return result.rowcount
 
+    @staticmethod
+    def get_single_value(result):
+        return result.scalar()
+
     def dispose_engine_after_use(func):
         """
         To dispose of engine

plugins/flows/data_transformation/dataflow_ui_plugin/fhirutils/utils.py

@@ -18,6 +18,7 @@ class omop_transform_utils:
     target_field_types = {
         "observation": {
             "observation_date": "date",
+            "observation_datetime": "datetime",
             "observation_id": "id",
             "person_id": "referenceToId",
         },

plugins/flows/data_transformation/dataflow_ui_plugin/nodes.py

@@ -38,6 +38,13 @@
 from _shared_flow_utils.api.SupabaseStorageAPI import SupabaseStorageAPI
 from subprocess import Popen, PIPE
 
+# Resolved relative to this module rather than hardcoded to /app/flows/... -
+# run-flow.sh stages each plugin under a per-run directory (e.g.
+# /var/lib/d2e-flows/data-transformation-flow/<sha>/flows/dataflow_ui_plugin),
+# not a fixed /app/flows path (/app only holds the worker's own scripts).
+_PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
 class Node:
     def __init__(self, name, node):
         self.id = node["id"]
@@ -468,12 +475,12 @@ def transform_fhir_data(self, input_fhir_df: pd.DataFrame = None) -> pd.DataFram
             raise Exception(f"OMOP table mapping not found for target structure definition url: {target_structure_definition_url}")
 
         source_resource_name = source_structure_definition_url.rstrip("/").split("/")[-1]
-        folder = "/app/flows/dataflow_ui_plugin/fhirutils/fhir_structureDefinition"
+        folder = os.path.join(_PLUGIN_DIR, "fhirutils", "fhir_structureDefinition")
         source_structure_definition = self.get_fhir_structure_definition(folder, source_resource_name)
         if not source_structure_definition:
             raise Exception(f"Source Structure Definition not found for url: {source_structure_definition_url}")
 
-        folder = "/app/flows/dataflow_ui_plugin/fhirutils/omop_structureDefinition"
+        folder = os.path.join(_PLUGIN_DIR, "fhirutils", "omop_structureDefinition")
         target_structure_definition = self.get_omop_structure_definition_by_url(folder, target_structure_definition_url)
 
         fhir_resource = None
@@ -482,7 +489,7 @@ def transform_fhir_data(self, input_fhir_df: pd.DataFrame = None) -> pd.DataFram
             fhir_resource = content_list if content_list else None
 
         transformed_omop = []
-        script_path = '/app/flows/dataflow_ui_plugin/fhirutils/fhir_transform.js'
+        script_path = os.path.join(_PLUGIN_DIR, "fhirutils", "fhir_transform.js")
 
         print("Starting FHIR Transform...")
         if fhir_resource:
@@ -606,22 +613,30 @@ def _truncate_table(self, dbconn, dialect: str) -> None:
     def task(self, _input: dict[str, Result], task_run_context):
         try:
             upstream = _input.get(self.dataframe)
-            if upstream is None or upstream.result is None or len(upstream.result) == 0:
-                return Result(True, f"No input data: the incoming dataframe from '{self.dataframe}' is empty", self, task_run_context)
+            if upstream is None:
+                return Result(True, f"No input data: upstream node '{self.dataframe}' did not run", self, task_run_context)
             if upstream.error:
                 return Result(True, f"No input data: upstream node '{self.dataframe}' failed, fix that node first", self, task_run_context)
-            if not isinstance(upstream.result, pd.DataFrame):
+            if upstream.result is not None and not isinstance(upstream.result, pd.DataFrame):
                 return Result(True, f"No input data: result from '{self.dataframe}' is not a dataframe", self, task_run_context)
-            df_to_write = upstream.result
 
             dbutils = DBDao(database_code=self.database)
             if dbutils.dialect == SupportedDatabaseDialects.TREX.value:
                 return Result(True, f"Writing to a trex database ('{self.database}') is not supported by the DB writer node", self, task_run_context)
             dbconn = dbutils.engine
 
+            # Truncate whenever the upstream ran validly, even with zero rows - it's a
+            # full-refresh semantics (empty source this run should mean an empty table,
+            # not a stale one left over from a previous run), distinct from an upstream
+            # that errored or never ran, which is left untouched above.
             if self.truncate:
                 self._truncate_table(dbconn, dbutils.dialect)
 
+            df_to_write = upstream.result
+            if df_to_write is None or len(df_to_write) == 0:
+                note = "Table truncated; " if self.truncate else ""
+                return Result(False, f"{note}no rows to write: the incoming dataframe from '{self.dataframe}' is empty", self, task_run_context)
+
             result = df_to_write.to_sql(
                 self.table_name,
                 dbconn,
@@ -1014,8 +1029,12 @@ def _ensure_mapping_schema(self, database_code: str, schema_name: str, dao: DBDa
         """)
 
         dao.execute_sql(f"""
-            CREATE UNIQUE INDEX IF NOT EXISTS fhir_omop_key_map_fhir_id_fhir_resource_type_idx
-            ON "{escaped_schema}".fhir_omop_key_map (fhir_id, fhir_resource_type)
+            DROP INDEX IF EXISTS "{escaped_schema}".fhir_omop_key_map_fhir_id_fhir_resource_type_idx
+        """)
+
+        dao.execute_sql(f"""
+            CREATE UNIQUE INDEX IF NOT EXISTS fhir_omop_key_map_fhir_id_type_table_omop_id_idx
+            ON "{escaped_schema}".fhir_omop_key_map (fhir_id, fhir_resource_type, omop_table_name, omop_id)
         """)
 
     def task(self, _input: dict[str, Result], task_run_context) -> Result:
@@ -1054,6 +1073,30 @@ def task(self, _input: dict[str, Result], task_run_context) -> Result:
             )
             omop_rows = list(omop_rows_df.itertuples(index=False, name=None))
 
+            # Reconcile rather than upsert-only: existing lineage rows for this
+            # (omop_table_name, fhir_resource_type) were written against whatever
+            # omop_id values the OMOP table had at that time. If the upstream
+            # DbWriter truncated and re-inserted that table since, the old omop_id
+            # values may no longer exist or may now belong to a different row -
+            # ON CONFLICT DO NOTHING on the insert below would never catch that,
+            # since a freshly-assigned omop_id doesn't collide with the stale one.
+            # Clearing unconditionally (before the empty-check, so a truncate-to-
+            # nothing run still clears out the now-orphaned old rows) keeps the
+            # mapping in sync with current OMOP contents whether this run's write
+            # was a full refresh or incremental.
+            mapping_escaped_schema = mapping_schema.replace('"', '""')
+            escaped_resource_type = self.fhir_resource_type.replace("'", "''")
+            escaped_omop_table_name = self.omop_table_name.replace("'", "''")
+            mapping_dao.execute_sql(f"""
+                DELETE FROM "{mapping_escaped_schema}".data_source
+                WHERE omop_table_name = '{escaped_omop_table_name}' AND fhir_resource_type = '{escaped_resource_type}'
+            """)
+            if self.write_key_map:
+                mapping_dao.execute_sql(f"""
+                    DELETE FROM "{mapping_escaped_schema}".fhir_omop_key_map
+                    WHERE omop_table_name = '{escaped_omop_table_name}' AND fhir_resource_type = '{escaped_resource_type}'
+                """)
+
             if not omop_rows:
                 return Result(False, {"inserted": 0, "updated": 0}, self, task_run_context)
 
@@ -1108,7 +1151,7 @@ def task(self, _input: dict[str, Result], task_run_context) -> Result:
                         )
                         for row in key_map_values
                     ],
-                    on_conflict="ON CONFLICT (fhir_id, fhir_resource_type) DO NOTHING",
+                    on_conflict="ON CONFLICT (fhir_id, fhir_resource_type, omop_table_name, omop_id) DO NOTHING",
                 )
 
             return Result(False, {"inserted": len(omop_rows), "updated": len(omop_rows)}, self, task_run_context)