From ef7c303f75254b150b525c6edddcce3225bec722 Mon Sep 17 00:00:00 2001 From: Afreen Sikandara Date: Fri, 26 Jun 2026 16:36:59 +0800 Subject: [PATCH] Fix syntax --- flows/etlQuestionnaireResponseEQ-5D-5L.json | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/flows/etlQuestionnaireResponseEQ-5D-5L.json b/flows/etlQuestionnaireResponseEQ-5D-5L.json index 0dd2fb4..2e531fa 100644 --- a/flows/etlQuestionnaireResponseEQ-5D-5L.json +++ b/flows/etlQuestionnaireResponseEQ-5D-5L.json @@ -75,7 +75,7 @@ "dataframe": "py2table", "schemaname": "dest_schema", "dbtablename": "measurement", - "description": "Insert the records into omop measurement table", + "description": "Insert the records into omop measurement table" }, "type": "db_writer_node", "width": 350, @@ -103,7 +103,7 @@ "path": "$", "source": "prepare_measurement_ids" }, - "description": "Describe the task of node py2table_node_0", + "description": "Describe the task of node py2table_node_0" }, "type": "py2table_node", "width": 350, @@ -127,7 +127,7 @@ "data": { "name": "normalize_and_assign_person_id", "description": "Convert column wise data to tabular", - "python_code": "# Input: transform_fhir_to_omop (dataframe) - OMOP measurement records output from the FHIR-to-OMOP transform node\n# Output: result (dataframe) - Measurement records with resolved person_id and cleaned value_as_number\n\nEQ_VAS_CONCEPT_ID = 42537274\n\n\ndef _unwrap_input_value(value):\n return value.result if hasattr(value, \"result\") else value\n\n\ndef extract_patient_ref(value):\n if isinstance(value, dict):\n return value.get(\"reference\") or value.get(\"id\") or str(value)\n return str(value) if value is not None else None\n\n\ndef normalize_fhir_id(value):\n if value is None:\n return None\n s = str(value).strip()\n if not s:\n return None\n if \"/\" in s:\n s = s.split(\"/\")[-1]\n return s\n\n\ndef load_person_id_map(database_code: str, schema_name: str) -> dict[str, int]:\n mapping_schema = f\"{database_code}_{schema_name}_fhir_mapping\"\n dao = DBDao(\n dialect=SupportedDatabaseDialects.TREX,\n database_code=database_code,\n )\n escaped_schema = mapping_schema.replace('\"', '\"\"')\n query = f\"\"\"\n SELECT fhir_id, omop_id\n FROM \"{escaped_schema}\".\"fhir_omop_key_map\"\n WHERE omop_table_name = 'person'\n AND omop_id IS NOT NULL\n \"\"\"\n rows = dao.execute_sql(query, fetch=True) or []\n result = {}\n for fhir_id, omop_id in rows:\n if fhir_id is None:\n continue\n try:\n omop_int = int(omop_id)\n except Exception:\n continue\n raw_key = str(fhir_id).strip()\n if raw_key:\n result[raw_key] = omop_int\n norm_key = normalize_fhir_id(fhir_id)\n if norm_key:\n result[norm_key] = omop_int\n print(f\"[fhir mapping] loaded {len(result)} person mappings from {mapping_schema}.fhir_omop_key_map\")\n return result\n\n\ndef exec(myinput):\n data = _unwrap_input_value(myinput.get(\"transform_fhir_to_omop\"))\n\n if data is None or data.empty:\n return pd.DataFrame()\n\n if not dest_db or not dest_schema:\n raise ValueError(\"Missing `dest_db` or `dest_schema`\")\n\n person_id_map = load_person_id_map(dest_db, dest_schema)\n if not person_id_map:\n raise ValueError(\n f\"No person mappings found in {dest_db}_{dest_schema}_fhir_mapping.fhir_omop_key_map — \"\n \"run the person pipeline before the measurement pipeline\"\n )\n\n print(f\"[normalize] columns: {list(data.columns)}, rows: {len(data)}\")\n\n result_rows = []\n skipped = []\n for _, row in data.iterrows():\n raw_ref = extract_patient_ref(row.get(\"person_id\"))\n normalized_ref = normalize_fhir_id(raw_ref)\n raw_ref_key = str(raw_ref).strip() if raw_ref is not None else None\n\n person_id = person_id_map.get(raw_ref_key) or person_id_map.get(normalized_ref)\n if person_id is None:\n skipped.append(raw_ref_key)\n continue\n\n entry = row.to_dict()\n entry[\"person_id\"] = person_id\n entry[\"value_as_number\"] = entry.get(\"value_as_number\") if entry.get(\"measurement_concept_id\") == EQ_VAS_CONCEPT_ID else None\n result_rows.append(entry)\n\n if skipped:\n print(f\"[normalize] skipped {len(skipped)} rows with no person mapping: {skipped[:10]}\")\n\n if not result_rows:\n raise ValueError(\n f\"All {len(data)} rows skipped — no person_id matches found in fhir_omop_key_map. \"\n f\"Sample unmatched refs: {skipped[:5]}\"\n )\n\n result_df = pd.DataFrame(result_rows)\n print(f\"[normalize] output shape: {result_df.shape}\")\n return result_df.reset_index(drop=True)\n", + "python_code": "# Input: transform_fhir_to_omop (dataframe) - OMOP measurement records output from the FHIR-to-OMOP transform node\n# Output: result (dataframe) - Measurement records with resolved person_id and cleaned value_as_number\n\nEQ_VAS_CONCEPT_ID = 42537274\n\n\ndef _unwrap_input_value(value):\n return value.result if hasattr(value, \"result\") else value\n\n\ndef extract_patient_ref(value):\n if isinstance(value, dict):\n return value.get(\"reference\") or value.get(\"id\") or str(value)\n return str(value) if value is not None else None\n\n\ndef normalize_fhir_id(value):\n if value is None:\n return None\n s = str(value).strip()\n if not s:\n return None\n if \"/\" in s:\n s = s.split(\"/\")[-1]\n return s\n\n\ndef load_person_id_map(database_code: str, schema_name: str) -> dict[str, int]:\n mapping_schema = f\"{database_code}_{schema_name}_fhir_mapping\"\n dao = DBDao(\n dialect=SupportedDatabaseDialects.TREX,\n database_code=database_code,\n )\n escaped_schema = mapping_schema.replace('\"', '\"\"')\n query = f\"\"\"\n SELECT fhir_id, omop_id\n FROM \"{escaped_schema}\".\"fhir_omop_key_map\"\n WHERE omop_table_name = 'person'\n AND omop_id IS NOT NULL\n \"\"\"\n rows = dao.execute_sql(query, fetch=True) or []\n result = {}\n for fhir_id, omop_id in rows:\n if fhir_id is None:\n continue\n try:\n omop_int = int(omop_id)\n except Exception:\n continue\n raw_key = str(fhir_id).strip()\n if raw_key:\n result[raw_key] = omop_int\n norm_key = normalize_fhir_id(fhir_id)\n if norm_key:\n result[norm_key] = omop_int\n print(f\"[fhir mapping] loaded {len(result)} person mappings from {mapping_schema}.fhir_omop_key_map\")\n return result\n\n\ndef exec(myinput):\n data = _unwrap_input_value(myinput.get(\"transform_fhir_to_omop\"))\n\n if data is None or data.empty:\n return pd.DataFrame()\n\n if not dest_db or not dest_schema:\n raise ValueError(\"Missing `dest_db` or `dest_schema`\")\n\n person_id_map = load_person_id_map(dest_db, dest_schema)\n if not person_id_map:\n raise ValueError(\n f\"No person mappings found in {dest_db}_{dest_schema}_fhir_mapping.fhir_omop_key_map — \"\n \"run the person pipeline before the measurement pipeline\"\n )\n\n print(f\"[normalize] columns: {list(data.columns)}, rows: {len(data)}\")\n\n result_rows = []\n skipped = []\n for _, row in data.iterrows():\n raw_ref = extract_patient_ref(row.get(\"person_id\"))\n normalized_ref = normalize_fhir_id(raw_ref)\n raw_ref_key = str(raw_ref).strip() if raw_ref is not None else None\n\n person_id = person_id_map.get(raw_ref_key) or person_id_map.get(normalized_ref)\n if person_id is None:\n skipped.append(raw_ref_key)\n continue\n\n entry = row.to_dict()\n entry[\"person_id\"] = person_id\n entry[\"value_as_number\"] = entry.get(\"value_as_number\") if entry.get(\"measurement_concept_id\") == EQ_VAS_CONCEPT_ID else None\n result_rows.append(entry)\n\n if skipped:\n print(f\"[normalize] skipped {len(skipped)} rows with no person mapping: {skipped[:10]}\")\n\n if not result_rows:\n raise ValueError(\n f\"All {len(data)} rows skipped — no person_id matches found in fhir_omop_key_map. \"\n f\"Sample unmatched refs: {skipped[:5]}\"\n )\n\n result_df = pd.DataFrame(result_rows)\n print(f\"[normalize] output shape: {result_df.shape}\")\n return result_df.reset_index(drop=True)\n" }, "type": "python_node", "width": 350, @@ -151,7 +151,7 @@ "data": { "name": "prepare_measurement_ids", "description": "Clean the OMOP CDM tables", - "python_code": "# Input: normalize_and_assign_person_id (dataframe) - Measurement records with resolved person_id from the normalize_and_assign_person_id node\n# Output: result (dataframe) - Measurement records with assigned measurement_id values, ready for db insert\n\ndef exec(myinput):\n data = myinput.get(\"normalize_and_assign_person_id\").result\n\n if data is None or data.empty:\n return data\n\n data = data.dropna(subset=[\"measurement_source_value\"])\n if data.empty:\n return data\n\n dao = SqlAlchemyDao(database_code=dest_db)\n fhir_ids = data[\"measurement_source_value\"].unique().tolist()\n\n omop_meta = sql.MetaData(schema=dest_schema)\n measurement_table = sql.Table(\"measurement\", omop_meta, autoload_with=dao.engine)\n\n # 1. Look up existing (source_value, concept_id) → measurement_id BEFORE deleting\n existing_rows = dao.execute_sqlalchemy_statement(\n sql.select(\n measurement_table.c.measurement_source_value,\n measurement_table.c.measurement_concept_id,\n measurement_table.c.measurement_id,\n ).where(measurement_table.c.measurement_source_value.in_(fhir_ids)),\n lambda r: r.fetchall(),\n ) or []\n existing = {(row[0], row[1]): row[2] for row in existing_rows}\n print(f\"Found {len(existing)} existing measurement rows to reuse\")\n\n # 2. Delete existing measurement rows for these FHIR IDs\n deleted = dao.execute_sqlalchemy_statement(\n measurement_table.delete().where(\n measurement_table.c.measurement_source_value.in_(fhir_ids)\n ),\n lambda r: r.rowcount,\n )\n print(f\"Deleted {deleted} rows from {dest_schema}.measurement\")\n\n # 3. Assign IDs — reuse by (source_value, concept_id), mint new for new records\n max_id = dao.execute_sqlalchemy_statement(\n sql.select(sql.func.coalesce(sql.func.max(measurement_table.c.measurement_id), 0)),\n lambda r: r.scalar(),\n )\n\n data = data.copy()\n next_id = max(max_id, max(existing.values(), default=0)) + 1\n assigned_ids = []\n for _, row in data.iterrows():\n key = (row[\"measurement_source_value\"], int(row[\"measurement_concept_id\"]))\n if key in existing:\n assigned_ids.append(existing[key])\n else:\n assigned_ids.append(next_id)\n next_id += 1\n\n data[\"measurement_id\"] = assigned_ids\n reused = sum(1 for i in assigned_ids if i in {v for v in existing.values()})\n print(f\"Assigned measurement_ids: {reused} reused, {len(assigned_ids) - reused} new\")\n return data\n", + "python_code": "# Input: normalize_and_assign_person_id (dataframe) - Measurement records with resolved person_id from the normalize_and_assign_person_id node\n# Output: result (dataframe) - Measurement records with assigned measurement_id values, ready for db insert\n\ndef exec(myinput):\n data = myinput.get(\"normalize_and_assign_person_id\").result\n\n if data is None or data.empty:\n return data\n\n data = data.dropna(subset=[\"measurement_source_value\"])\n if data.empty:\n return data\n\n dao = SqlAlchemyDao(database_code=dest_db)\n fhir_ids = data[\"measurement_source_value\"].unique().tolist()\n\n omop_meta = sql.MetaData(schema=dest_schema)\n measurement_table = sql.Table(\"measurement\", omop_meta, autoload_with=dao.engine)\n\n # 1. Look up existing (source_value, concept_id) → measurement_id BEFORE deleting\n existing_rows = dao.execute_sqlalchemy_statement(\n sql.select(\n measurement_table.c.measurement_source_value,\n measurement_table.c.measurement_concept_id,\n measurement_table.c.measurement_id,\n ).where(measurement_table.c.measurement_source_value.in_(fhir_ids)),\n lambda r: r.fetchall(),\n ) or []\n existing = {(row[0], row[1]): row[2] for row in existing_rows}\n print(f\"Found {len(existing)} existing measurement rows to reuse\")\n\n # 2. Delete existing measurement rows for these FHIR IDs\n deleted = dao.execute_sqlalchemy_statement(\n measurement_table.delete().where(\n measurement_table.c.measurement_source_value.in_(fhir_ids)\n ),\n lambda r: r.rowcount,\n )\n print(f\"Deleted {deleted} rows from {dest_schema}.measurement\")\n\n # 3. Assign IDs — reuse by (source_value, concept_id), mint new for new records\n max_id = dao.execute_sqlalchemy_statement(\n sql.select(sql.func.coalesce(sql.func.max(measurement_table.c.measurement_id), 0)),\n lambda r: r.scalar(),\n )\n\n data = data.copy()\n next_id = max(max_id, max(existing.values(), default=0)) + 1\n assigned_ids = []\n for _, row in data.iterrows():\n key = (row[\"measurement_source_value\"], int(row[\"measurement_concept_id\"]))\n if key in existing:\n assigned_ids.append(existing[key])\n else:\n assigned_ids.append(next_id)\n next_id += 1\n\n data[\"measurement_id\"] = assigned_ids\n reused = sum(1 for i in assigned_ids if i in {v for v in existing.values()})\n print(f\"Assigned measurement_ids: {reused} reused, {len(assigned_ids) - reused} new\")\n return data\n" }, "type": "python_node", "width": 350,