Feature/1414 add rail transport dataset

[jh-RLI]

Fixes #1414

Adds a new dataset RailTransitDemand that spatially and temporally allocates Germany's rail-transport electricity demand onto the eTraGo grid model as loads, building on eGon's own OSM data. Three carriers are covered: 16.7-Hz traction (rail_traction), DC tram/U-Bahn (rail_transit_dc) and DC S-Bahn (rail_sbahn_dc).

What it does

• Coupling points from eGon's OSM tables: DC rectifier Unterwerke are extracted and classified from openstreetmap.osm_point / osm_ways / osm_polygon (via hstore(tags)->'power'/'voltage'/'frequency'); 16.7-Hz converter locations come from a curated list in the data bundle (OSM under-tags them).
• Spatial allocation: each city's DC energy is distributed over its nearby OSM rectifiers (≤25 km, else city centroid); loads are bus-assigned by coupling level (HöS/HS → EHV substation voronoi, MS → MV grid district, nearest-neighbour fallback).
• Time series in MW, weather year 2011: the normalized hourly profiles (from the bundle) are re-indexed from 2025 to the eGon weather year 2011 (the weekday order differs), then scaled to absolute MW — p_set[h] = energy_mwh_a · factor(scn) · profile_2011[h].
• Scenarios via scenario_parameters: introduces three reGon scenarios — status2024, reGon2037, reGon2045 — with the gross rail demand stored as mobility → rail_transport_demand → annual_demand (MWh, eGon unit convention). status2024 uses the absolute series 1:1; the futures scale by total(scn)/total(status2024), derived from the NEP figures (Schiene+Bus combined; bus excluded as the ~1.3 TWh remainder to NEP's 14 TWh). Only the data that eGon cannot derive itself (load profiles, GTFS-based energy weights, curated converter list) lives in the data bundle.

Notes on scope/assumptions: the NEP does not split rail vs. bus, so the future growth uses the combined NEP ratio as a rail-growth proxy (documented in code). The reGon scenarios currently carry only the mobility sector; they are intended to grow into full scenarios later.

📋 Pull Request Guidelines

Please read the Pull Request Guidelines carefully before creating your PR.

---

🧑‍💻 Contributor Checklist

Before requesting a review, make sure you've completed all of the following:

• All tests pass locally or via CI
  (for more information on local test, check tox in the Contributing section)
  (CI tests are automatically executed when creating a PR, you can see the results of the checks below)
• Workflow has run at least once in Test mode
  (optional if no dataset changes are involved)
• Relevant documentation is updated (API, new features, etc.)
  (module/dataset docstrings; project docs maintained externally)
• Dataset-versions are updated when existing datasets are adjusted.
  (RailTransitDemand 0.0.3, ScenarioParameters 0.0.21 → 0.0.22)
• Added a note to CHANGELOG.rst about the changes
• Added yourself to AUTHORS.rst

Optional:

• Changes have been tested in Everything mode
• Extend the checklist for reviewers: Which aspects should be reviewed in particular?

Please focus on:
- The MW unit handling + 2011 re-indexing in `_profiles_2011`
  (each profile column sums to 1, so energy_mwh_a · profile[h] = MW).
- The per-scenario scaling in `insert_rail_demand`
  (factor = total(scn)/total(status2024); unit-invariant ratio).
- The OSM DC-rectifier classification (`_is_dc_rectifier`) and the
  25 km city→rectifier attachment / centroid fallback.
- Bus assignment by grid level in `_assign_bus`.
- Creating the three reGon scenarios in `insert_scenarios` (mobility-only
  for now; other sectors empty).

---

🔍 Reviewer Checklist

During your review, please check the following:

• Is the code clean, readable, and efficient? Are there any oddities or obvious inefficiencies?
• Does the code work as expected? (should already be verified by contributor)
• Do all tests pass? (see CI results)
• Is the documentation complete and up to date?
• Is CHANGELOG.rst updated accordingly?
• Is all necessary metadata complete and correct?
  • If metadata is pending: Is there an appropriate issue filed?

---

📝 Additional Notes (optional)

• DB test run still pending. The dataset has not yet been run against the egon DB; to verify after osm + mv_grid_districts + substation_voronoi + setup_etrago + scenario_parameters, trigger the dataset and check:

SELECT carrier, count(*) FROM grid.egon_etrago_load
WHERE carrier LIKE 'rail_%' GROUP BY carrier;

and confirm scenario.egon_scenario_parameters contains status2024 / reGon2037 / reGon2045.

• Data bundle dependency: the dataset reads data_bundle_egon_data/rail_transport_demand/ (load profiles + GTFS energy weights + curated converter list).
• Two scenario_parameters files carry pre-existing flake8 violations (unrelated to this change); only the newly added lines were modified.

[jh-RLI]

@nesnoj

I will keep my mini / test DAGS in the .gitignore for now and remove them once we can merge this PR.

Is this issue with the CI already kown? The pipeline fails for and it states that the DB is not availabale. Below claude inpected the issue and found a possible cause:

Error from Tox setp:

  .tox/py310-nocov-linux/lib/python3.10/site-packages/psycopg2/__init__.py:122: in connect
      conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
  E   psycopg2.OperationalError: connection to server at "127.0.0.1", port 59734 failed: Connection refused
  E   	Is the server running on that host and accepting TCP/IP connections?

Quick claude analysis

Diagnosis
It's not a flaky/unavailable DB — it's a code regression that makes DAG construction require a live database. And it's on dev, so it breaks CI for the whole repo, not just your feature/1414 branch.

Root cause: The recently merged feature "Store datasets sources and targets in separate table" (36e4238) added a self.register() call to Dataset.post_init. So now, constructing any Dataset object immediately opens a PostgreSQL connection:

src/egon/data/datasets/init.py:453

def register(self):
    SourcesTargetsModel.__table__.create(bind=db.engine(), checkfirst=True)  # <-- live connection
    with db.session_scope() as session:
        ...

The two failing tests never needed a DB before:

test_pipeline_importability just imports the pipeline — but importing builds every Dataset, which now hits the DB.
test_uniqueness_of_automatically_generated_final_dataset_task just instantiates two Datasets.
The CI workflow (build.yml) has no Postgres service, so db.engine() connects to 127.0.0.1: and gets Connection refused. (test_airflow, which really needs a DB, is explicitly skipped — confirming CI was never meant to have Postgres.)

There's a deeper smell too: Airflow re-parses DAG files constantly, so requiring a DB connection just to import the DAG is fragile even in production.

Recommended fix
Make register() tolerant of a missing DB — skip registration with a warning instead of crashing. Registration still happens normally during real pipeline runs (where the DB exists); imports, DAG parsing, and unit tests stop depending on it.

def register(self):
    try:
        SourcesTargetsModel.__table__.create(bind=db.engine(), checkfirst=True)
        with db.session_scope() as session:
            ...  # existing body
    except OperationalError as e:
        logger.warning(
            f"Could not register sources/targets for '{self.name}' "
            f"(database unavailable): {e}. Skipping registration."
        )

This is minimal and low-risk. The alternative — adding a postgres service container to build.yml — is more invasive, slower, and the test suite clearly wasn't designed to assume one.

Want me to apply the tolerant-register() fix? I'd do it on your current branch (feature/1414-add-rail-transport-dataset), though note the underlying bug lives on dev, so ideally the fix lands there too.

[jh-RLI]

@nesnoj The Codacy errors are false flags IMO - ofc its not good to use f string with user input to construct a SQL query but here we dont have any user input but the defined input from sources / targets

[jh-RLI]

This is the "mini" DAG that aims to isolate only to relevant datasets related to the new RailTransitDemand dataset. It runs sucessfully:

Mini DAG for Testing RailTransitDemand dataset

"""Isolated DAG: build every upstream dataset RailTransitDemand needs, then
run it -- so the reGon rail transport demand can be tested end-to-end against
a freshly filled DB, without the whole eGon pipeline.

Runs the transitive dependency closure of RailTransitDemand:
  setup, setup_etrago, osm, data_bundle, vg250, scenario_parameters,
  substation_extraction, osmtgmod, substation_voronoi, mv_grid_districts,
  then rail_transit_demand.

The TaskGroup ids ("input_data", "electricity_grid") mirror pipeline.py so the
fully-qualified task references inside Osmtgmod / SubstationVoronoi resolve.
NOTE: osmtgmod is heavy (it builds the ehv/hv grid model). Gitignored dev DAG.
"""

import os

from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup
import airflow

from egon.data.config import set_numexpr_threads
from egon.data.config import settings as egon_settings
from egon.data.datasets import database
from egon.data.datasets.data_bundle import DataBundle
from egon.data.datasets.etrago_setup import EtragoSetup
from egon.data.datasets.mv_grid_districts import mv_grid_districts_setup
from egon.data.datasets.osm import OpenStreetMap
from egon.data.datasets.osmtgmod import Osmtgmod
from egon.data.datasets.rail_transport_demand import RailTransitDemand
from egon.data.datasets.sanity_checks import (
    sanitycheck_rail_transport_demand,
)
from egon.data.datasets.scenario_parameters import ScenarioParameters
from egon.data.datasets.substation import SubstationExtraction
from egon.data.datasets.substation_voronoi import SubstationVoronoi
from egon.data.datasets.vg250 import Vg250

set_numexpr_threads()

prefix = egon_settings()["egon-data"].get("--prefix")
prefix = "" if prefix is None else f"{prefix}-"

with airflow.DAG(
    f"{prefix}egon-rail-transport-isolated",
    description="Isolated DAG: rail transport demand + its upstream chain.",
    default_args={"start_date": days_ago(1)},
    template_searchpath=[
        os.path.abspath(
            os.path.join(
                os.path.dirname(__file__), "..", "..", "processing", "vg250"
            )
        )
    ],
    is_paused_upon_creation=False,
    schedule_interval=None,
) as rail_pipeline:
    tasks = rail_pipeline.task_dict

    with TaskGroup(group_id="setup") as setup_group:
        setup = database.Setup()
        setup_etrago = EtragoSetup(dependencies=[setup])

    with TaskGroup(group_id="input_data") as input_data_group:
        osm = OpenStreetMap(dependencies=[setup])
        data_bundle = DataBundle(dependencies=[setup])
        vg250 = Vg250(dependencies=[setup])
        scenario_parameters = ScenarioParameters(dependencies=[setup])

    with TaskGroup(group_id="electricity_grid") as electricity_grid_group:
        substation_extraction = SubstationExtraction(
            dependencies=[osm, vg250]
        )
        osmtgmod = Osmtgmod(
            dependencies=[
                scenario_parameters,
                setup_etrago,
                substation_extraction,
                tasks["input_data.osm.download"],
            ]
        )
        substation_voronoi = SubstationVoronoi(
            dependencies=[
                tasks["electricity_grid.osmtgmod.substation.extract"],
                vg250,
            ]
        )
        mv_grid_districts = mv_grid_districts_setup(
            dependencies=[substation_voronoi]
        )

    with TaskGroup(group_id="mobility_demand") as mobility_demand_group:
        rail_transit_demand = RailTransitDemand(
            dependencies=[
                data_bundle,
                osm,
                mv_grid_districts,
                substation_voronoi,
                setup_etrago,
                scenario_parameters,
            ]
        )

    with TaskGroup(group_id="sanity_checks") as sanity_checks_group:
        # Validate the rail loads/timeseries the dataset just wrote. Only the
        # rail check runs here (the full SanityChecks suite needs other
        # scenarios/datasets not built by this isolated DAG).
        rail_sanity_check = PythonOperator(
            task_id="sanitycheck_rail_transport_demand",
            python_callable=sanitycheck_rail_transport_demand,
        )

    for last_task in rail_transit_demand.tasks.last:
        last_task.set_downstream(rail_sanity_check)