Merge branch 'dev' into features/#381-hh-electricity-demand-per-bus

.github/pull_request_template.md

@@ -0,0 +1,11 @@
+Fixes # . 
+
+## Before merging into `dev`-branch, please make sure that
+
+- [ ] the `CHANGELOG.rst` was updated.
+- [ ] new and adjusted code is formated using `black` and `isort`.
+- [ ] the `Dataset`-version is updated when existing datasets are adjusted. 
+- [ ] the branch was merged into the `continuous-integration/run-everything-over-the-weekend`- branch.
+- [ ] the workflow is running successful in `test mode`.
+- [ ] the workflow is running successful in `Everything` mode.
+

CHANGELOG.rst

@@ -157,6 +157,12 @@ Added
   `#485 <https://github.com/openego/eGon-data/issues/485>`_
 * Include time-dependent coefficient of performance for heat pumps
   `#532 <https://github.com/openego/eGon-data/issues/532>`_
+* Limit number of parallel processes per task
+  `#265 <https://github.com/openego/eGon-data/issues/265>`_
+* Include biomass CHP plants to eTraGo tables
+  `#498 <https://github.com/openego/eGon-data/issues/498>`_
+* Include feedin time series for wind offshore
+  `#531 <https://github.com/openego/eGon-data/issues/531>`_
 
 .. _PR #159: https://github.com/openego/eGon-data/pull/159
 
@@ -244,12 +250,16 @@ Changed
   `#484 <https://github.com/openego/eGon-data/issues/484>`_
 * Migrate dlr script to datasets
   `#508 <https://github.com/openego/eGon-data/issues/508>`_
+* Migrate loadarea scripts to datasets
+  `#525 <https://github.com/openego/eGon-data/issues/525>`_
 * Migrate plot.py to dataset of district heating areas
   `#527 <https://github.com/openego/eGon-data/issues/527>`_
 * Migrate substation scripts to datasets
   `#304 <https://github.com/openego/eGon-data/issues/304>`_
 * Add household demand profiles to etrago table
   `#381 <https://github.com/openego/eGon-data/issues/381>`_
+* Migrate zensus scripts to datasets
+  `#422 <https://github.com/openego/eGon-data/issues/422>`_
 
 
 Bug fixes

src/egon/data/airflow/dags/pipeline.py

@@ -7,6 +7,7 @@
 import importlib_resources as resources
 
 from egon.data import db
+from egon.data.config import set_numexpr_threads
 from egon.data.datasets import database
 from egon.data.datasets.calculate_dlr import Calculate_dlr
 from egon.data.datasets.ch4_storages import CH4Storages
@@ -39,6 +40,7 @@
 from egon.data.datasets.industrial_gas_demand import IndustrialGasDemand
 from egon.data.datasets.industrial_sites import MergeIndustrialSites
 from egon.data.datasets.industry import IndustrialDemandCurves
+from egon.data.datasets.loadarea import LoadArea
 from egon.data.datasets.mastr import mastr_data_setup
 from egon.data.datasets.mv_grid_districts import mv_grid_districts_setup
 from egon.data.datasets.osm import OpenStreetMap
@@ -54,17 +56,19 @@
 from egon.data.datasets.substation_voronoi import SubstationVoronoi
 from egon.data.datasets.vg250 import Vg250
 from egon.data.datasets.vg250_mv_grid_districts import Vg250MvGridDistricts
+from egon.data.datasets.zensus import ZensusPopulation, ZensusMiscellaneous
 from egon.data.datasets.zensus_mv_grid_districts import ZensusMvGridDistricts
 from egon.data.datasets.zensus_vg250 import ZensusVg250
 from egon.data.processing.gas_areas import GasAreas
 from egon.data.processing.power_to_h2 import PowertoH2
 import egon.data.datasets.gas_grid as gas_grid
-import egon.data.importing.zensus as import_zs
 import egon.data.processing.gas_areas as gas_areas
-import egon.data.processing.loadarea as loadarea
 import egon.data.processing.power_to_h2 as power_to_h2
 
 
+# Set number of threads used by numpy and pandas
+set_numexpr_threads()
+
 with airflow.DAG(
     "egon-data-processing-pipeline",
     description="The eGo^N data processing DAG.",
@@ -104,41 +108,16 @@
     scenario_parameters = ScenarioParameters(dependencies=[setup])
     scenario_input_import = tasks["scenario_parameters.insert-scenarios"]
 
-    # Zensus import
-    zensus_download_population = PythonOperator(
-        task_id="download-zensus-population",
-        python_callable=import_zs.download_zensus_pop,
-    )
-
-    zensus_download_misc = PythonOperator(
-        task_id="download-zensus-misc",
-        python_callable=import_zs.download_zensus_misc,
-    )
-
-    zensus_tables = PythonOperator(
-        task_id="create-zensus-tables",
-        python_callable=import_zs.create_zensus_tables,
-    )
-
-    population_import = PythonOperator(
-        task_id="import-zensus-population",
-        python_callable=import_zs.population_to_postgres,
-    )
-
-    zensus_misc_import = PythonOperator(
-        task_id="import-zensus-misc",
-        python_callable=import_zs.zensus_misc_to_postgres,
-    )
-    setup >> zensus_download_population >> zensus_download_misc
-    zensus_download_misc >> zensus_tables >> population_import
-    vg250_clean_and_prepare >> population_import
-    population_import >> zensus_misc_import
+    # Zensus population import
+    zensus_population = ZensusPopulation(dependencies=[setup, vg250])
 
     # Combine Zensus and VG250 data
-    zensus_vg250 = ZensusVg250(dependencies=[vg250, population_import])
-    zensus_inside_ger = tasks["zensus_vg250.inside-germany"]
+    zensus_vg250 = ZensusVg250(dependencies=[vg250, zensus_population])
 
-    zensus_inside_ger >> zensus_misc_import
+    # Download and import zensus data on households, buildings and apartments
+    zensus_miscellaneous = ZensusMiscellaneous(
+        dependencies=[zensus_population, zensus_vg250]
+    )
 
     # DemandRegio data import
     demandregio = DemandRegio(
@@ -148,20 +127,15 @@
 
     # Society prognosis
     society_prognosis = SocietyPrognosis(
-        dependencies=[
-            demandregio,
-            zensus_vg250,
-            population_import,
-            zensus_misc_import,
-        ]
+        dependencies=[demandregio, zensus_vg250, zensus_population]
     )
 
     # Distribute household electrical demands to zensus cells
     household_electricity_demand_annual = HouseholdElectricityDemand(
         dependencies=[
             demandregio,
             zensus_vg250,
-            zensus_tables,
+            zensus_miscellaneous,
             society_prognosis,
         ]
     )
@@ -172,11 +146,8 @@
 
     # NEP data import
     scenario_capacities = ScenarioCapacities(
-        dependencies=[setup, vg250, data_bundle]
+        dependencies=[setup, vg250, data_bundle, zensus_population]
     )
-    nep_insert_data = tasks["scenario_capacities.insert-data-nep"]
-
-    population_import >> nep_insert_data
 
     # setting etrago input tables
 
@@ -194,22 +165,15 @@
 
     # osmTGmod ehv/hv grid model generation
     osmtgmod = Osmtgmod(
-        dependencies=[
-            osm_download,
-            substation_extraction,
-            setup_etrago,
-        ]
+        dependencies=[osm_download, substation_extraction, setup_etrago]
     )
     osmtgmod.insert_into(pipeline)
     osmtgmod_pypsa = tasks["osmtgmod.to-pypsa"]
     osmtgmod_substation = tasks["osmtgmod_substation"]
 
     # create Voronoi polygons
     substation_voronoi = SubstationVoronoi(
-        dependencies=[
-            osmtgmod_substation,
-            vg250,
-        ]
+        dependencies=[osmtgmod_substation, vg250]
     )
 
     # MV grid districts
@@ -259,21 +223,7 @@
     )
 
     # Extract landuse areas from osm data set
-    create_landuse_table = PythonOperator(
-        task_id="create-landuse-table",
-        python_callable=loadarea.create_landuse_table,
-    )
-
-    landuse_extraction = PostgresOperator(
-        task_id="extract-osm_landuse",
-        sql=resources.read_text(loadarea, "osm_landuse_extraction.sql"),
-        postgres_conn_id="egon_data",
-        autocommit=True,
-    )
-    setup >> create_landuse_table
-    create_landuse_table >> landuse_extraction
-    osm_add_metadata >> landuse_extraction
-    vg250_clean_and_prepare >> landuse_extraction
+    load_area = LoadArea(dependencies=[osm, vg250])
 
     # Import weather data
     weather_data = WeatherData(
@@ -289,12 +239,14 @@
 
     # District heating areas demarcation
     district_heating_areas = DistrictHeatingAreas(
-        dependencies=[heat_demand_Germany, scenario_parameters]
+        dependencies=[
+            heat_demand_Germany,
+            scenario_parameters,
+            zensus_miscellaneous,
+        ]
     )
     import_district_heating_areas = tasks["district_heating_areas.demarcation"]
 
-    zensus_misc_import >> import_district_heating_areas
-
     # Calculate dynamic line rating for HV trans lines
     dlr = Calculate_dlr(
         dependencies=[
@@ -306,7 +258,7 @@
 
     # Map zensus grid districts
     zensus_mv_grid_districts = ZensusMvGridDistricts(
-        dependencies=[population_import, mv_grid_districts]
+        dependencies=[zensus_population, mv_grid_districts]
     )
 
     map_zensus_grid_districts = tasks["zensus_mv_grid_districts.mapping"]
@@ -348,9 +300,9 @@
     hh_demand = hh_demand_setup(
         dependencies=[
             vg250_clean_and_prepare,
-            zensus_misc_import,
+            zensus_miscellaneous,
             map_zensus_grid_districts,
-            zensus_inside_ger,
+            zensus_vg250,
             demandregio,
         ],
         tasks=(
@@ -378,7 +330,7 @@
             industrial_sites,
             demandregio_demand_cts_ind,
             osm,
-            landuse_extraction,
+            load_area,
         ]
     )
 
@@ -405,7 +357,6 @@
     chp_locations_nep = tasks["chp.insert-chp-egon2035"]
     chp_heat_bus = tasks["chp.assign-heat-bus"]
 
-    nep_insert_data >> chp_locations_nep
     import_district_heating_areas >> chp_locations_nep
 
     # Power plants
@@ -439,7 +390,8 @@
 
     # Fill eTraGo Generators tables
     fill_etrago_generators = Egon_etrago_gen(
-        dependencies=[power_plants, weather_data])
+        dependencies=[power_plants, weather_data]
+    )
 
     # Heat supply
     heat_supply = HeatSupply(
@@ -454,7 +406,12 @@
 
     # Heat to eTraGo
     heat_etrago = HeatEtrago(
-        dependencies=[heat_supply, mv_grid_districts, setup_etrago, renewable_feedin]
+        dependencies=[
+            heat_supply,
+            mv_grid_districts,
+            setup_etrago,
+            renewable_feedin,
+        ]
     )
 
     heat_etrago_buses = tasks["heat_etrago.buses"]

src/egon/data/cli.py

@@ -109,6 +109,17 @@
     ),
     show_default=True,
 )
+@click.option(
+    "--processes-per-task",
+    default=1,
+    metavar="N_PROCESS",
+    help=(
+        "Each task can use at maximum N_PROCESS parallel processes. Remember"
+        " that in addition to that, tasks can run in parallel (see N) and"
+        " there's always the scheduler and probably the serverrunning."
+    ),
+    show_default=True,
+)
 @click.option(
     "--docker-container-name",
     default="egon-data-local-database-container",

src/egon/data/config.py

@@ -102,3 +102,17 @@ def datasets(config_file=None):
         config_file = os.path.join(package_path, "datasets.yml")
 
     return yaml.load(open(config_file), Loader=yaml.SafeLoader)
+
+def set_numexpr_threads():
+    """Sets maximum threads used by NumExpr
+
+    Returns
+    -------
+    None
+
+    """
+    # Read maximum number of threads per task from egon-data.configuration.yaml
+    num_processes = settings()["egon-data"]["--processes-per-task"]
+
+    os.environ['NUMEXPR_MAX_THREADS'] = str(num_processes)
+    os.environ['NUMEXPR_NUM_THREADS'] = str(num_processes)

src/egon/data/datasets.yml

@@ -710,6 +710,7 @@ chp_location:
     vg250_lan:
       schema: 'boundaries'
       table: 'vg250_lan'
+    mastr_biomass: "bnetza_mastr_biomass_cleaned.csv"
   targets:
     chp_table:
       table: 'egon_chp_plants'
@@ -733,6 +734,9 @@ chp_etrago:
     link:
       schema: 'grid'
       table: 'egon_etrago_link'
+    generator:
+      schema: 'grid'
+      table: 'egon_etrago_generator'
 
 DSM_CTS_industry:
   sources: