Calculate index value - #3295
Open
csafreen wants to merge 4 commits into
Open
Conversation
csafreen
marked this pull request as ready for review
September 9, 2026 14:10
csafreen
requested review from
SantanM,
Zhimin-arya,
alicia-koh,
brandantck,
mwaiyee,
p-hoffmann and
suwarnoong
as code owners
September 9, 2026 14:10
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Runtime provisioning, parameter schema, transactional replacement, ID allocation, and lineage consistency issues remain unresolved.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds an EQ-5D-5L Prefect flow that calculates country-specific utility indices from OMOP observations and writes measurements with FHIR lineage.
Changes:
- Adds scoring logic, country value sets, configuration, documentation, and unit tests.
- Integrates the flow into worker packaging and CI.
- Updates shared database and FHIR mapping behavior.
File summaries
| File | Description |
|---|---|
services/alp-dataflow-gen-worker/Dockerfile |
Packages and provisions the new flow. |
plugins/flows/eq5d5l_index_calculation/pyproject.toml |
Defines Python and Pixi dependencies. |
plugins/flows/eq5d5l_index_calculation/package.json |
Registers the flow and parameter schema. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/types.py |
Defines configuration and concept constants. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/tests/test_scoring.py |
Tests scoring behavior. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/scoring.py |
Implements value-set parsing and scoring. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/README.md |
Documents usage and data contracts. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/flow.py |
Reads observations and writes calculated measurements. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/flow.json |
Adds a supporting FHIR transformation flow. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Ghana.txt |
Adds Ghana coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Germany.txt |
Adds Germany coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/France.txt |
Adds France coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Ethiopia.txt |
Adds Ethiopia coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Egypt.txt |
Adds Egypt coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Denmark.txt |
Adds Denmark coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/China.txt |
Adds China coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Canada.txt |
Adds Canada coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Belgium.txt |
Adds Belgium coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/external/value_sets/Australia.txt |
Adds Australia coefficients. |
plugins/flows/eq5d5l_index_calculation/eq5d5l_index_calculation_plugin/__init__.py |
Creates the plugin package. |
plugins/flows/data_transformation/dataflow_ui_plugin/nodes.py |
Updates resource paths, empty writes, and lineage reconciliation. |
plugins/flows/_shared_flow_utils/dao/sqlalchemydao.py |
Adds selection and atomic replacement helpers. |
docker-compose-local.yml |
Enables the plugin-specific local override. |
.github/workflows/docker-build-push.yaml |
Adds the plugin to image builds. |
.github/workflows/_pixi-lock-check.yml |
Adds lockfile validation. |
Review details
- Files reviewed: 24/26 changed files
- Comments generated: 7
- Review effort level: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+279
to
+284
| 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) |
| "name": "prepare_measurement_ids", | ||
| "error": false, | ||
| "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", |
Comment on lines
+273
to
+278
| mapping_dao.batch_insert_values( | ||
| mapping_schema, | ||
| "fhir_omop_key_map", | ||
| ["fhir_id", "fhir_resource_type", "omop_table_name", "omop_id"], | ||
| key_map_rows, | ||
| on_conflict="ON CONFLICT (fhir_id, fhir_resource_type, omop_table_name, omop_id) DO NOTHING", |
| RUN --mount=type=cache,target=/opt/pixi/cache,sharing=locked \ | ||
| set -e; \ | ||
| for group in cohort-discovery-flow loyalty-score-flow i2b2-flow \ | ||
| for group in loyalty-score-flow eq5d5l-index-calculation-flow i2b2-flow \ |
Comment on lines
+114
to
+123
| def _extract_level(code: Optional[str], answer_code_level_map: Optional[dict]) -> Optional[int]: | ||
| if code is None: | ||
| return None | ||
| try: | ||
| return int(code) | ||
| except (TypeError, ValueError): | ||
| pass | ||
| if answer_code_level_map and code in answer_code_level_map: | ||
| return answer_code_level_map[code] | ||
| return None |
Comment on lines
+59
to
+64
| "cache_id": { | ||
| "anyOf": [ | ||
| { "type": "string" }, | ||
| { "type": "null" } | ||
| ], | ||
| "title": "Cache Id" |
Comment on lines
+86
to
+91
| - ./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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Merge Checklist
Please cross check this list if additions / modifications needs to be done on top of your core changes and tick them off. Reviewer can as well glance through and help the developer if something is missed out.
developbranch)