Update metadata integration to oemetadata v2

.gitignore

@@ -69,3 +69,10 @@ docs/_build
 
 # Mypy Cache
 .mypy_cache/
+
+# oemetadata
+/src/egon/data/metadata/results/converted
+
+# Dev Airflow DAGS for testing
+/src/egon/data/airflow/dags/mini_dag_zensu.py
+/src/egon/data/airflow/dags/pipeline_mini_dag_metadata.py

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

@@ -3,8 +3,8 @@
 from airflow.utils.dates import days_ago
 import airflow
 
-from egon.data.config import settings as egon_settings
 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.ch4_prod import CH4Production
 from egon.data.datasets.ch4_storages import CH4Storages
@@ -45,7 +45,9 @@
 from egon.data.datasets.heat_etrago import HeatEtrago
 from egon.data.datasets.heat_etrago.hts_etrago import HtsEtragoTable
 from egon.data.datasets.heat_supply import HeatSupply
-from egon.data.datasets.heat_supply.individual_heating import HeatPumpsStatusQuo
+from egon.data.datasets.heat_supply.individual_heating import (
+    HeatPumpsStatusQuo,
+)
 from egon.data.datasets.industrial_sites import MergeIndustrialSites
 from egon.data.datasets.industry import IndustrialDemandCurves
 from egon.data.datasets.loadarea import LoadArea, OsmLanduse
@@ -81,9 +83,11 @@
 with airflow.DAG(
     f"{prefix}powerd-status-quo-processing-pipeline",
     description="The PoWerD Status Quo data processing DAG.",
-    default_args={"start_date": days_ago(1),
-                  "email_on_failure": False,
-                  "email":"clara.buettner@hs-flensburg.de"},
+    default_args={
+        "start_date": days_ago(1),
+        "email_on_failure": False,
+        "email": "clara.buettner@hs-flensburg.de",
+    },
     template_searchpath=[
         os.path.abspath(
             os.path.join(
@@ -203,8 +207,8 @@
     # TODO: What does "trans" stand for?
     # Calculate dynamic line rating for HV (high voltage) trans lines
     # dlr = Calculate_dlr(
-    #    dependencies=[data_bundle, osmtgmod, weather_data] # , fix_subnetworks]
-    #)
+    #    dependencies=[data_bundle, osmtgmod, weather_data] # , fix_subnetworks] # noqa: E501
+    # )
 
     # Map zensus grid districts
     zensus_mv_grid_districts = ZensusMvGridDistricts(
@@ -363,7 +367,12 @@
     )
     # Create gas voronoi status quo
     create_gas_polygons_statusquo = GasAreas(
-        dependencies=[setup_etrago, vg250, gas_grid_insert_data, substation_voronoi]
+        dependencies=[
+            setup_etrago,
+            vg250,
+            gas_grid_insert_data,
+            substation_voronoi,
+        ]
     )
 
     # Gas abroad
@@ -437,7 +446,6 @@
         ]
     )
 
-
     # Pumped hydro and home storage units
     storage_units = Storages(
         dependencies=[
@@ -531,5 +539,3 @@
             demand_curves_industry,
         ]
     )
-
-

src/egon/data/datasets/DSM_cts_ind.py

@@ -12,7 +12,6 @@
 import datetime
 import json
 
-from omi.dialects import get_dialect
 from sqlalchemy import ARRAY, Column, Float, Integer, String
 from sqlalchemy.ext.declarative import declarative_base
 import geopandas as gpd
@@ -28,7 +27,6 @@
     contributors,
     generate_resource_fields_from_db_table,
     license_odbl,
-    meta_metadata,
     sources,
 )
 
@@ -368,39 +366,9 @@ def add_metadata_individual():
                 }
             ],
             "review": {"path": "", "badge": ""},
-            "metaMetadata": meta_metadata(),
-            "_comment": {
-                "metadata": (
-                    "Metadata documentation and explanation (https://"
-                    "github.com/OpenEnergyPlatform/oemetadata/blob/master/"
-                    "metadata/v141/metadata_key_description.md)"
-                ),
-                "dates": (
-                    "Dates and time must follow the ISO8601 including time "
-                    "zone (YYYY-MM-DD or YYYY-MM-DDThh:mm:ss±hh)"
-                ),
-                "units": "Use a space between numbers and units (100 m)",
-                "languages": (
-                    "Languages must follow the IETF (BCP47) format (en-GB, "
-                    "en-US, de-DE)"
-                ),
-                "licenses": (
-                    "License name must follow the SPDX License List "
-                    "(https://spdx.org/licenses/)"
-                ),
-                "review": (
-                    "Following the OEP Data Review (https://github.com/"
-                    "OpenEnergyPlatform/data-preprocessing/wiki)"
-                ),
-                "none": "If not applicable use (none)",
-            },
         }
 
-        dialect = get_dialect(
-            f"oep-v{meta_metadata()['metadataVersion'][4:7]}"
-        )()
-
-        meta = dialect.compile_and_render(dialect.parse(json.dumps(meta)))
+        meta = json.dumps(meta)
 
         db.submit_comment(
             f"'{json.dumps(meta)}'",

src/egon/data/datasets/chp/__init__.py

@@ -1,9 +1,8 @@
 """
-The central module containing all code dealing with combined heat and power
-(CHP) plants.
+The central module containing all code dealing with combined heat and
+power (CHP) plants.
 """
 
-from pathlib import Path
 import datetime
 import json
 import time
@@ -16,7 +15,6 @@
 from sqlalchemy.orm import sessionmaker
 import geopandas as gpd
 import pandas as pd
-import pypsa
 
 from egon.data import config, db
 from egon.data.datasets import Dataset, wrapped_partial
@@ -43,7 +41,6 @@
     context,
     generate_resource_fields_from_sqla_model,
     license_egon_data_odbl,
-    meta_metadata,
     sources,
 )
 
@@ -95,26 +92,26 @@ def metadata():
     fields_df = pd.DataFrame(data=fields).set_index("name")
     fields_df.loc["id", "description"] = "Unique identifyer"
     fields_df.loc["sources", "description"] = "List of sources"
-    fields_df.loc["source_id", "description"] = (
-        "Names of sources, e.g. MaStr_id"
-    )
+    fields_df.loc[
+        "source_id", "description"
+    ] = "Names of sources, e.g. MaStr_id"
     fields_df.loc["carrier", "description"] = "Energy carrier"
-    fields_df.loc["district_heating", "description"] = (
-        "Used in district heating or not"
-    )
-    fields_df.loc["el_capacity", "description"] = (
-        "Installed electrical capacity"
-    )
+    fields_df.loc[
+        "district_heating", "description"
+    ] = "Used in district heating or not"
+    fields_df.loc[
+        "el_capacity", "description"
+    ] = "Installed electrical capacity"
     fields_df.loc["th_capacity", "description"] = "Installed thermal capacity"
-    fields_df.loc["electrical_bus_id", "description"] = (
-        "Index of corresponding electricity bus"
-    )
-    fields_df.loc["district_heating_area_id", "description"] = (
-        "Index of corresponding district heating bus"
-    )
-    fields_df.loc["ch4_bus_id", "description"] = (
-        "Index of corresponding methane bus"
-    )
+    fields_df.loc[
+        "electrical_bus_id", "description"
+    ] = "Index of corresponding electricity bus"
+    fields_df.loc[
+        "district_heating_area_id", "description"
+    ] = "Index of corresponding district heating bus"
+    fields_df.loc[
+        "ch4_bus_id", "description"
+    ] = "Index of corresponding methane bus"
     fields_df.loc["voltage_level", "description"] = "Voltage level"
     fields_df.loc["scenario", "description"] = "Name of scenario"
     fields_df.loc["geom", "description"] = "Location of CHP plant"
@@ -167,10 +164,12 @@ def metadata():
                     "primaryKey": ["index"],
                     "foreignKeys": [],
                 },
-                "dialect": {"delimiter": None, "decimalSeparator": "."},
+                "dialect": {
+                    "delimiter": None,
+                    "decimalSeparator": ".",
+                },
             }
         ],
-        "metaMetadata": meta_metadata(),
     }
 
     # Add metadata as a comment to the table
@@ -329,7 +328,7 @@ def assign_heat_bus():
                     district_heating=row.district_heating,
                     voltage_level=row.voltage_level,
                     scenario=scenario,
-                    geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
+                    geom=(f"SRID=4326;POINT({row.geom.x} {row.geom.y})"),
                 )
             else:
                 entry = EgonChp(
@@ -344,7 +343,7 @@ def assign_heat_bus():
                     district_heating=row.district_heating,
                     voltage_level=row.voltage_level,
                     scenario=scenario,
-                    geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
+                    geom=(f"SRID=4326;POINT({row.geom.x} {row.geom.y})"),
                 )
             session.add(entry)
         session.commit()
@@ -365,7 +364,7 @@ def insert_biomass_chp(scenario):
     """
     cfg = config.datasets()["chp_location"]
 
-    # import target values from NEP 2021, scneario C 2035
+    # import target values from NEP 2021, scenario C 2035
     target = select_target("biomass", scenario)
 
     # import data for MaStR
@@ -386,7 +385,7 @@ def insert_biomass_chp(scenario):
         )
     ]
 
-    # Scaling will be done per federal state in case of eGon2035 scenario.
+    # Scaling per federal state in case of eGon2035 scenario.
     if scenario == "eGon2035":
         level = "federal_state"
     else:
@@ -423,7 +422,7 @@ def insert_biomass_chp(scenario):
                 district_heating=row.district_heating,
                 electrical_bus_id=row.bus_id,
                 voltage_level=row.voltage_level,
-                geom=f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})",
+                geom=(f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})"),
             )
             session.add(entry)
     session.commit()
@@ -521,7 +520,8 @@ def insert_chp_statusquo(scn="status2019"):
 
     print(
         f"""
-          CHPs with a total installed electrical capacity of {dropped_capacity} kW are dropped
+          CHPs with a total installed electrical
+          capacity of {dropped_capacity} kW are dropped
           because of missing or wrong location data
           """
     )
@@ -562,7 +562,7 @@ def insert_chp_statusquo(scn="status2019"):
                 electrical_bus_id=row.bus_id,
                 ch4_bus_id=row.gas_bus_id,
                 voltage_level=row.voltage_level,
-                geom=f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})",
+                geom=(f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})"),
             )
             session.add(entry)
     session.commit()
@@ -674,7 +674,8 @@ def extension_SH():
 
 
 def insert_chp_egon100re():
-    """Insert CHP plants for eGon100RE considering results from pypsa-eur-sec
+    """Insert CHP plants for eGon100RE considering results from
+    pypsa-eur-sec
 
     Returns
     -------
@@ -817,28 +818,40 @@ class Chp(Dataset):
     """
     Extract combined heat and power plants for each scenario
 
-    This dataset creates combined heat and power (CHP) plants for each scenario and defines their use case.
-    The method bases on existing CHP plants from Marktstammdatenregister. For the eGon2035 scenario,
-    a list of CHP plans from the grid operator is used for new largescale CHP plants. CHP < 10MW are
-    randomly distributed.
-    Depending on the distance to a district heating grid, it is decided if the CHP is used to
-    supply a district heating grid or used by an industrial site.
+    This dataset creates combined heat and power (CHP) plants for each
+    scenario and defines their use case. The method bases on existing CHP
+    plants from Marktstammdatenregister. For the eGon2035 scenario, a list of
+    CHP plans from the grid operator is used for new largescale CHP plants.
+    CHP < 10MW are randomly distributed. Depending on the distance to a
+    district heating grid, it is decided if the CHP is used to supply a
+    district heating grid or used by an industrial site.
 
 
     *Dependencies*
-      * :py:class:`GasAreaseGon100RE <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
-      * :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
-      * :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
-      * :py:class:`IndustrialDemandCurves <egon.data.datasets.industry.IndustrialDemandCurves>`
-      * :py:class:`OsmLanduse <egon.data.datasets.loadarea.OsmLanduse>`
-      * :py:func:`download_mastr_data <egon.data.datasets.mastr.download_mastr_data>`
-      * :py:func:`define_mv_grid_districts <egon.data.datasets.mv_grid_districts.define_mv_grid_districts>`
-      * :py:class:`ScenarioCapacities <egon.data.datasets.scenario_capacities.ScenarioCapacities>`
+      * :py:class:`GasAreaseGon100RE
+        <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
+      * :py:class:`GasAreaseGon2035
+        <egon.data.datasets.gas_areas.GasAreaseGon2035>`
+      * :py:class:`DistrictHeatingAreas
+        <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
+      * :py:class:`IndustrialDemandCurves
+        <egon.data.datasets.industry.IndustrialDemandCurves>`
+      * :py:class:`OsmLanduse
+        <egon.data.datasets.loadarea.OsmLanduse>`
+      * :py:func:`download_mastr_data
+        <egon.data.datasets.mastr.download_mastr_data>`
+      * :py:func:`define_mv_grid_districts
+        <egon.data.datasets.mv_grid_districts.define_mv_grid_districts>`
+      * :py:class:`ScenarioCapacities
+        <egon.data.datasets.scenario_capacities.ScenarioCapacities>`
 
 
     *Resulting tables*
-      * :py:class:`supply.egon_chp_plants <egon.data.datasets.chp.EgonChp>` is created and filled
-      * :py:class:`supply.egon_mastr_conventional_without_chp <egon.data.datasets.chp.EgonMaStRConventinalWithoutChp>` is created and filled
+      * :py:class:`supply.egon_chp_plants
+        <egon.data.datasets.chp.EgonChp>` is created and filled
+      * :py:class:`supply.egon_mastr_conventional_without_chp
+        <egon.data.datasets.chp.EgonMaStRConventinalWithoutChp>` is created
+        and filled
 
     """