Conversation
There was a problem hiding this comment.
Pull request overview
This PR migrates the simulator worker execution path to a Prefect-based flow/deployment model while modernizing development tooling (uv + ruff + ty), CI, and container build strategy.
Changes:
- Introduces a new Prefect flow (
simulator_flow) and deployment script, plus supporting env helpers and local-run entrypoint. - Migrates tooling/config: switches to uv build backend and dependency groups, replaces flake8/mypy with ruff/ty, and updates CI + just tasks accordingly.
- Updates/expands tests and fixtures to exercise the Prefect flow path and deployment job variables.
Reviewed changes
Copilot reviewed 43 out of 46 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| unit_test/test_hello.py | Reformats legacy unittest file (now duplicated by tests/test_hello.py). |
| tests/test_prefect_flow.py | Adds a Prefect flow execution test with MinIO/InfluxDB mocking. |
| tests/test_prefect_deploy_flow.py | Adds coverage for module-level job_variables structure. |
| tests/test_kpi_integration.py | Reworks KPI integration to call Prefect flow and handle Prefect State. |
| tests/test_hello.py | Adds new test suite for config parsing/date indexing under tests/. |
| tests/data/esdl/simulator_tutorial.esdl | Adds ESDL fixture used by Prefect flow tests/local run. |
| src/simulator_worker/worker_task.py | Adds a legacy worker task module, currently fully commented out. |
| src/simulator_worker/utils.py | Refactors utilities typing/logging; adds _parse_datetime_config; tweaks ESDL output naming. |
| src/simulator_worker/simulator_worker.py | Removes previous main worker entrypoint implementation. |
| src/simulator_worker/prefect_flow.py | Adds Prefect flow implementation and artifact writing to MinIO. |
| src/simulator_worker/prefect_deploy_flow.py | Adds Prefect deployment registration script and job variables. |
| src/simulator_worker/env.py | Adds environment access helpers (require_env, EnvSettings). |
| src/simulator_worker/main.py | Removes previous __main__ entrypoint. |
| src/simulator_worker/init.py | Changes init behavior to load dotenv (removes logging + script export). |
| run.sh | Removes old local run script. |
| run.ps1 | Removes old PowerShell helper script. |
| requirements.txt | Removes pip-compile generated runtime requirements. |
| README.md | Updates dev workflow docs for uv/just/ruff/ty and Prefect deploy/run steps. |
| pyproject.toml | Switches to uv build backend, updates deps, and replaces flake8/mypy config with ruff/ty-era config. |
| local_run/run_simulator_flow_function.py | Adds local runner for calling the Prefect flow function without Prefect orchestration. |
| justfile | Adds standardized local/CI task runner commands. |
| Dockerfile | Reworks container build to use uv and BuildKit mounts; removes previous entrypoint-based install flow. |
| dev.Dockerfile | Reworks dev image build to use uv and allow local SDK install. |
| dev-requirements.txt | Removes pip-compile generated dev requirements. |
| ci/win32/update_dependencies.cmd | Removes legacy Windows dependency update script. |
| ci/win32/typecheck.cmd | Removes legacy Windows mypy script. |
| ci/win32/test_unit.cmd | Removes legacy Windows unittest/pytest script. |
| ci/win32/lint.cmd | Removes legacy Windows flake8 script. |
| ci/win32/install_dependencies.cmd | Removes legacy Windows pip-sync script. |
| ci/win32/create_venv.cmd | Removes legacy Windows venv bootstrap script. |
| ci/linux/update_dependencies.sh | Removes legacy Linux dependency update script. |
| ci/linux/typecheck.sh | Removes legacy Linux mypy script. |
| ci/linux/test_unit.sh | Removes legacy Linux pytest runner script (unit_test-based). |
| ci/linux/lint.sh | Removes legacy Linux flake8 script. |
| ci/linux/install_dependencies.sh | Removes legacy Linux pip-sync script. |
| ci/linux/create_venv.sh | Removes legacy Linux venv bootstrap script. |
| ci/linux/build_python_package.sh | Removes legacy package build script. |
| ci/linux/_load_dot_env.sh | Removes legacy dotenv loader script. |
| .vscode/settings.json | Updates editor integration for ruff/ty; switches test discovery to tests/. |
| .vscode/launch.json | Adds debug configs for Prefect deploy, local flow run, and pytest. |
| .gitignore | Renames coverage ignore path from unit_test_coverage/ to tests_coverage/. |
| .github/workflows/publish_container_image.yml | Modernizes GHCR publish workflow using buildx + metadata-action. |
| .github/workflows/ci.yml | Replaces venv-tar pipeline with uv+just jobs (lint/security/test/typecheck). |
| .env.template.local | Updates local env template for Prefect/MinIO/InfluxDB settings. |
| .env.template | Updates env template for Prefect/MinIO/InfluxDB settings. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| if minio_host is None or minio_port is None or minio_access_key is None or minio_secret is None: | ||
| raise ValueError( | ||
| f"MinIO credentials are not fully set. MinIO host: {minio_host}, port: {minio_port}, " | ||
| f"access key: {minio_access_key}, secret key: {minio_secret}" | ||
| ) |
| value = config[key] | ||
| if not isinstance(value, str): | ||
| return default | ||
|
|
||
| try: | ||
| return datetime.fromisoformat(value) | ||
| except ValueError: | ||
| return default |
| Returns: | ||
| SimulatorFlowResult | State[Any] | None: Failed state when execution fails; otherwise no value is | ||
| returned from this flow function. |
| # """Main python file for Simulator-worker.""" | ||
| # import logging | ||
| # import math | ||
| # import traceback | ||
| # from datetime import datetime, timedelta |
jarsarasty
left a comment
jarsarasty
left a comment
There was a problem hiding this comment.
Nice work on the migration overall - the flow/deploy split is clean and EnvSettings is a big improvement over scattered os.getenv calls. A few things I'd like fixed before merge, plus one optional cleanup. Left them inline.
One coordination note that isn't a defect in this PR: the Rosim JAR download lands in this image (uv sync installs omotes-simulator-core==0.0.30 from PyPI), which is where the KeyError: 'assets' in the omotes-system system tests comes from. That's fixed upstream in simulator-core and will need a pin bump here once released. Be aware the ATES tests also fail on main through a different mechanism (a 120s timeout at wait_until_result, no KeyError), so the pin bump is necessary but may not be sufficient.
| output_esdl_with_kpis = build_esdl_string_with_kpis(output_esdl, kpi_results) | ||
| except Exception: | ||
| logging.exception("KPI calculation failed. Results will be returned without KPIs.") | ||
| output_esdl_with_kpis = output_esdl |
There was a problem hiding this comment.
Please fix: this makes a KPI failure invisible. The except Exception logs and assigns output_esdl_with_kpis = output_esdl, then execution continues unbroken to the artifact write and return success_result - so the flow reports success with KPIs silently missing, and nothing downstream inspects the ESDL for them.
That matters because test__simulator__kpis_present_in_output in omotes-system asserts exact CAPEX/OPEX values, so a regression here surfaces as a confusing system-test failure rather than a clear worker error.
Suggestion: keep the broad catch (a KPI failure arguably shouldn't sink an otherwise-good simulation) but surface it in esdl_messages instead of only the log. That field is declared on SimulatorFlowResult and currently hardcoded to [] at both return sites, so it's an existing channel that carries nothing today.
optimizer-worker/src/omotes_optimizer_worker/prefect_flow.py already does exactly this - collects list[EsdlMessage] and serialises with [m.model_dump(mode="json") for m in esdl_messages]. Same shape would work here:
except Exception as kpi_error:
logging.exception("KPI calculation failed. Results will be returned without KPIs.")
output_esdl_with_kpis = output_esdl
esdl_messages.append(
EsdlMessage(
technical_message=f"KPI calculation failed: {kpi_error}",
severity=MessageSeverity.WARNING,
)
)(from omotes_sdk.esdl_messages import EsdlMessage, MessageSeverity, then pass the dumped list into SimulatorFlowResult rather than [].) That way the run still succeeds, but the caller can tell KPIs are absent and why.
|
|
||
| # Debug output: save ESDL files if enabled (can be controlled via workflow_config) | ||
| # debug_enabled = _parse_bool_config(workflow_config, "debug_esdl", False) | ||
| debug_enabled = False |
There was a problem hiding this comment.
Please fix: debug_enabled = False is hardcoded, so the if debug_enabled: branch below is statically unreachable - save_debug_esdl (imported at the top) and the debug_esdl_dir read are dead code.
This one has a real consequence rather than just being untidy: tests/test_kpi_integration.py sets config["debug_esdl"] = True and config["debug_esdl_dir"] = debug_dir in setUpClass, and its class docstring says "It enables debug_esdl so the output ESDL is saved as a CI artifact for inspection." That's now a silent no-op - the temp dir is created, cleaned up, and nothing is ever written to it.
Either restore the commented-out _parse_bool_config line above, or drop the debug path and update that test's docstring so it doesn't claim something it no longer does.
| workflow_type_name: Name of the workflow. | ||
|
|
||
| Returns: | ||
| SimulatorFlowResult | State[Any] | None: Failed state when execution fails; otherwise no value is |
There was a problem hiding this comment.
Please fix: the docstring says "otherwise no value is returned from this flow function", but the body ends with return success_result. Worth correcting since this is the module's main entry point.
| logging.info(f"workflow config: {workflow_config}") | ||
| # timestep = workflow_config.get("timestep", 3600) # default to 1 hour in seconds | ||
| # start = datetime.fromisoformat(workflow_config.get("start_time", "2019-01-01T00:00:00+00:00")) | ||
| # end = datetime.fromisoformat(workflow_config.get("end_time", "2019-01-31T01:00:00+00:00")) |
There was a problem hiding this comment.
Please fix: these three commented-out lines can go - git history has them if they're ever needed.
| # timestep = workflow_config.get("timestep", 3600) # default to 1 hour in seconds | ||
| # start = datetime.fromisoformat(workflow_config.get("start_time", "2019-01-01T00:00:00+00:00")) | ||
| # end = datetime.fromisoformat(workflow_config.get("end_time", "2019-01-31T01:00:00+00:00")) | ||
| timestep = workflow_config["timestep"] |
There was a problem hiding this comment.
Cleanup, your call: timestep is the only config value read as a bare dict index while start_time/end_time go through _parse_datetime_config with defaults.
Not a bug - workflow_config_nwn.json declares timestep with "default": 3600, so the orchestrator always supplies it, and a KeyError would be caught by the outer handler into a Failed state anyway. Slightly funny that the config gives timestep a default while start_time/end_time have none, which is the opposite of how they're treated here. Just flagging the inconsistency.
| │ │ ├── prefect_flow.py # Prefect flow entry point for simulator runs | ||
| │ │ ├── py.typed # Marker for typed package consumers | ||
| │ │ ├── utils.py # ESDL/profile/KPI support utilities | ||
| │ │ └── worker_task.py # Legacy worker task module |
There was a problem hiding this comment.
Please fix: worker_task.py doesn't exist on this branch (or on main), so the structure diagram documents a module that isn't there.
No description provided.