data.datasets.storages.home_batteries.get_cbat_pbat_ratio() - Code Metrics - Inspection of "Merge pull request #568 from openego/features/#450..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 7cf077...0e9721 )

unknown

created 2025-04-16 13:04 UTC

get_cbat_pbat_ratio() A

↳ Parent: data.datasets.storages.home_batteries

Complexity

Conditions

Size

Total Lines	21
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	21
rs	10
c	0
b	0
f	0
cc	1
nop	0

"""
Home Battery allocation to buildings

Main module for allocation of home batteries onto buildings and sizing them
depending on pv rooftop system size.

**Contents of this module**
* Creation of DB tables
* Allocate given home battery capacity per mv grid to buildings with pv rooftop
  systems. The sizing of the home battery system depends on the size of the
  pv rooftop system and can be set within the *datasets.yml*. Default sizing is
  1:1 between the pv rooftop capacity (kWp) and the battery capacity (kWh).
* Write results to DB

**Configuration**

The config of this dataset can be found in *datasets.yml* in section
*home_batteries*.

**Scenarios and variations**

Assumptions can be changed within the *datasets.yml*.

Only buildings with a pv rooftop systems are considered within the allocation
process. The default sizing of home batteries is 1:1 between the pv rooftop
capacity (kWp) and the battery capacity (kWh). Reaching the exact value of the
allocation of battery capacities per grid area leads to slight deviations from
this specification.

## Methodology

The selection of buildings is done randomly until a result is reached which is
close to achieving the sizing specification.
"""
import datetime
import json

from loguru import logger
from numpy.random import RandomState
from omi.dialects import get_dialect
from sqlalchemy import Column, Float, Integer, String
from sqlalchemy.ext.declarative import declarative_base
import numpy as np
import pandas as pd

from egon.data import config, db
from egon.data.metadata import (
    context,
    contributors,
    generate_resource_fields_from_db_table,
    license_dedl,
    license_odbl,
    meta_metadata,
    meta_metadata,
    sources,
)

Base = declarative_base()


def get_cbat_pbat_ratio():
    """
    Mean ratio between the storage capacity and the power of the pv rooftop
    system

    Returns
    -------
    int
        Mean ratio between the storage capacity and the power of the pv
        rooftop system
    """
    sources = config.datasets()["home_batteries"]["sources"]

    sql = f"""
    SELECT max_hours
    FROM {sources["etrago_storage"]["schema"]}
    .{sources["etrago_storage"]["table"]}
    WHERE carrier = 'home_battery'
    """

    return int(db.select_dataframe(sql).iat[0, 0])


def allocate_home_batteries_to_buildings():
    """
    Allocate home battery storage systems to buildings with pv rooftop systems
    """
    # get constants
    constants = config.datasets()["home_batteries"]["constants"]
    scenarios = constants["scenarios"]
    cbat_ppv_ratio = constants["cbat_ppv_ratio"]
    rtol = constants["rtol"]
    max_it = constants["max_it"]
    cbat_pbat_ratio = get_cbat_pbat_ratio()

    sources = config.datasets()["home_batteries"]["sources"]

    df_list = []

    for scenario in scenarios:
        # get home battery capacity per mv grid id
        sql = f"""
        SELECT el_capacity as p_nom_min, bus_id as bus FROM
        {sources["storage"]["schema"]}
        .{sources["storage"]["table"]}
        WHERE carrier = 'home_battery'
        AND scenario = '{scenario}';
        """

        home_batteries_df = db.select_dataframe(sql)

        home_batteries_df = home_batteries_df.assign(
            bat_cap=home_batteries_df.p_nom_min * cbat_pbat_ratio
        )

        sql = """
        SELECT building_id, capacity
        FROM supply.egon_power_plants_pv_roof_building
        WHERE scenario = '{}'
        AND bus_id = {}
        """

        for bus_id, bat_cap in home_batteries_df[
            ["bus", "bat_cap"]
        ].itertuples(index=False):
            pv_df = db.select_dataframe(sql.format(scenario, bus_id))

            grid_ratio = bat_cap / pv_df.capacity.sum()

            if grid_ratio > cbat_ppv_ratio:
                logger.warning(
                    f"In Grid {bus_id} and scenario {scenario}, the ratio of "
                    f"home storage capacity to pv rooftop capacity is above 1"
                    f" ({grid_ratio: g}). The storage capacity of pv rooftop "
                    f"systems will be high."
                )

            if grid_ratio < cbat_ppv_ratio:
                random_state = RandomState(seed=bus_id)

                n = max(int(len(pv_df) * grid_ratio), 1)

                best_df = pv_df.sample(n=n, random_state=random_state)

                i = 0

                while (
                    not np.isclose(best_df.capacity.sum(), bat_cap, rtol=rtol)
                    and i < max_it
                ):
                    sample_df = pv_df.sample(n=n, random_state=random_state)

                    if abs(best_df.capacity.sum() - bat_cap) > abs(
                        sample_df.capacity.sum() - bat_cap
                    ):
                        best_df = sample_df.copy()

                    i += 1

                    if sample_df.capacity.sum() < bat_cap:
                        n = min(n + 1, len(pv_df))
                    else:
                        n = max(n - 1, 1)

                if not np.isclose(best_df.capacity.sum(), bat_cap, rtol=rtol):
                    logger.warning(
                        f"No suitable generators could be found in Grid "
                        f"{bus_id} and scenario {scenario} to achieve the "
                        f"desired ratio between battery capacity and pv "
                        f"rooftop capacity. The ratio will be "
                        f"{bat_cap / best_df.capacity.sum()}."
                    )

                pv_df = best_df.copy()

            bat_df = pv_df.drop(columns=["capacity"]).assign(
                capacity=pv_df.capacity / pv_df.capacity.sum() * bat_cap,
                p_nom=pv_df.capacity
                / pv_df.capacity.sum()
                * bat_cap
                / cbat_pbat_ratio,
                scenario=scenario,
                bus_id=bus_id,
            )

            df_list.append(bat_df)

    create_table(pd.concat(df_list, ignore_index=True))

    add_metadata()


class EgonHomeBatteries(Base):
    targets = config.datasets()["home_batteries"]["targets"]

    __tablename__ = targets["home_batteries"]["table"]
    __table_args__ = {"schema": targets["home_batteries"]["schema"]}

    index = Column(Integer, primary_key=True, index=True)
    scenario = Column(String)
    bus_id = Column(Integer)
    building_id = Column(Integer)
    p_nom = Column(Float)
    capacity = Column(Float)


def add_metadata():
    """
    Add metadata to table supply.egon_home_batteries
    """
    targets = config.datasets()["home_batteries"]["targets"]
    deposit_id_mastr = config.datasets()["mastr_new"]["deposit_id"]
    deposit_id_data_bundle = config.datasets()["data-bundle"]["sources"][
        "zenodo"
    ]["deposit_id"]

    contris = contributors(["kh", "kh"])

    contris[0]["date"] = "2023-03-15"

    contris[0]["object"] = "metadata"
    contris[1]["object"] = "dataset"

    contris[0]["comment"] = "Add metadata to dataset."
    contris[1]["comment"] = "Add workflow to generate dataset."

    meta = {
        "name": (
            f"{targets['home_batteries']['schema']}."
            f"{targets['home_batteries']['table']}"
        ),
        "title": "eGon Home Batteries",
        "id": "WILL_BE_SET_AT_PUBLICATION",
        "description": "Home storage systems allocated to buildings",
        "language": "en-US",
        "keywords": ["battery", "batteries", "home", "storage", "building"],
        "publicationDate": datetime.date.today().isoformat(),
        "context": context(),
        "spatial": {
            "location": "none",
            "extent": "Germany",
            "resolution": "building",
        },
        "temporal": {
            "referenceDate": "2021-12-31",
            "timeseries": {},
        },
        "sources": [
            {
                "title": "Data bundle for egon-data",
                "description": (
                    "Data bundle for egon-data: A transparent and "
                    "reproducible data processing pipeline for energy "
                    "system modeling"
                ),
                "path": (
                    "https://zenodo.org/record/"
                    f"{deposit_id_data_bundle}#.Y_dWM4CZMVM"
                ),
                "licenses": [license_dedl(attribution="© Cußmann, Ilka")],
            },
            {
                "title": ("open-MaStR power unit registry for eGo^n project"),
                "description": (
                    "Data from Marktstammdatenregister (MaStR) data using "
                    "the data dump from 2022-11-17 for eGon-data."
                ),
                "path": (
                    f"https://zenodo.org/record/{deposit_id_mastr}"
                ),
                "licenses": [license_dedl(attribution="© Amme, Jonathan")],
            },
            sources()["openstreetmap"],
            sources()["era5"],
            sources()["vg250"],
            sources()["egon-data"],
            sources()["nep2021"],
            sources()["mastr"],
            sources()["technology-data"],
        ],
        "licenses": [license_odbl("© eGon development team")],
        "contributors": contris,
        "resources": [
            {
                "profile": "tabular-data-resource",
                "name": (
                    f"{targets['home_batteries']['schema']}."
                    f"{targets['home_batteries']['table']}"
                ),
                "path": "None",
                "format": "PostgreSQL",
                "encoding": "UTF-8",
                "schema": {
                    "fields": generate_resource_fields_from_db_table(
                        targets["home_batteries"]["schema"],
                        targets["home_batteries"]["table"],
                    ),
                    "primaryKey": "index",
                },
                "dialect": {"delimiter": "", "decimalSeparator": ""},
            }
        ],
        "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(meta_metadata())()

    meta = dialect.compile_and_render(dialect.parse(json.dumps(meta)))

    db.submit_comment(
        f"'{json.dumps(meta)}'",
        targets["home_batteries"]["schema"],
        targets["home_batteries"]["table"],
    )


def create_table(df):
    """Create mapping table home battery <-> building id"""
    engine = db.engine()

    EgonHomeBatteries.__table__.drop(bind=engine, checkfirst=True)
    EgonHomeBatteries.__table__.create(bind=engine, checkfirst=True)

    df.reset_index().to_sql(
        name=EgonHomeBatteries.__table__.name,
        schema=EgonHomeBatteries.__table__.schema,
        con=engine,
        if_exists="append",
        index=False,
    )


1			"""
2			Home Battery allocation to buildings
3
4			Main module for allocation of home batteries onto buildings and sizing them
5			depending on pv rooftop system size.
6
7			Contents of this module
8			* Creation of DB tables
9			* Allocate given home battery capacity per mv grid to buildings with pv rooftop
10			systems. The sizing of the home battery system depends on the size of the
11			pv rooftop system and can be set within the datasets.yml. Default sizing is
12			1:1 between the pv rooftop capacity (kWp) and the battery capacity (kWh).
13			* Write results to DB
14
15			Configuration
16
17			The config of this dataset can be found in datasets.yml in section
18			home_batteries.
19
20			Scenarios and variations
21
22			Assumptions can be changed within the datasets.yml.
23
24			Only buildings with a pv rooftop systems are considered within the allocation
25			process. The default sizing of home batteries is 1:1 between the pv rooftop
26			capacity (kWp) and the battery capacity (kWh). Reaching the exact value of the
27			allocation of battery capacities per grid area leads to slight deviations from
28			this specification.
29
30			## Methodology
31
32			The selection of buildings is done randomly until a result is reached which is
33			close to achieving the sizing specification.
34			"""
35			import datetime
36			import json
37
38			from loguru import logger
39			from numpy.random import RandomState
40			from omi.dialects import get_dialect
41			from sqlalchemy import Column, Float, Integer, String
42			from sqlalchemy.ext.declarative import declarative_base
43			import numpy as np
44			import pandas as pd
45
46			from egon.data import config, db
47			from egon.data.metadata import (
48			context,
49			contributors,
50			generate_resource_fields_from_db_table,
51			license_dedl,
52			license_odbl,
53			meta_metadata,
54			meta_metadata,
55			sources,
56			)
57
58			Base = declarative_base()
59
60
61			def get_cbat_pbat_ratio():
62			"""
63			Mean ratio between the storage capacity and the power of the pv rooftop
64			system
65
66			Returns
67			-------
68			int
69			Mean ratio between the storage capacity and the power of the pv
70			rooftop system
71			"""
72			sources = config.datasets()["home_batteries"]["sources"]
73
74			sql = f"""
75			SELECT max_hours
76			FROM {sources["etrago_storage"]["schema"]}
77			.{sources["etrago_storage"]["table"]}
78			WHERE carrier = 'home_battery'
79			"""
80
81			return int(db.select_dataframe(sql).iat[0, 0])
82
83
84			def allocate_home_batteries_to_buildings():
85			"""
86			Allocate home battery storage systems to buildings with pv rooftop systems
87			"""
88			# get constants
89			constants = config.datasets()["home_batteries"]["constants"]
90			scenarios = constants["scenarios"]
91			cbat_ppv_ratio = constants["cbat_ppv_ratio"]
92			rtol = constants["rtol"]
93			max_it = constants["max_it"]
94			cbat_pbat_ratio = get_cbat_pbat_ratio()
95
96			sources = config.datasets()["home_batteries"]["sources"]
97
98			df_list = []
99
100			for scenario in scenarios:
101			# get home battery capacity per mv grid id
102			sql = f"""
103			SELECT el_capacity as p_nom_min, bus_id as bus FROM
104			{sources["storage"]["schema"]}
105			.{sources["storage"]["table"]}
106			WHERE carrier = 'home_battery'
107			AND scenario = '{scenario}';
108			"""
109
110			home_batteries_df = db.select_dataframe(sql)
111
112			home_batteries_df = home_batteries_df.assign(
113			bat_cap=home_batteries_df.p_nom_min * cbat_pbat_ratio
114			)
115
116			sql = """
117			SELECT building_id, capacity
118			FROM supply.egon_power_plants_pv_roof_building
119			WHERE scenario = '{}'
120			AND bus_id = {}
121			"""
122
123			for bus_id, bat_cap in home_batteries_df[
124			["bus", "bat_cap"]
125			].itertuples(index=False):
126			pv_df = db.select_dataframe(sql.format(scenario, bus_id))
127
128			grid_ratio = bat_cap / pv_df.capacity.sum()
129
130			if grid_ratio > cbat_ppv_ratio:
131			logger.warning(
132			f"In Grid {bus_id} and scenario {scenario}, the ratio of "
133			f"home storage capacity to pv rooftop capacity is above 1"
134			f" ({grid_ratio: g}). The storage capacity of pv rooftop "
135			f"systems will be high."
136			)
137
138			if grid_ratio < cbat_ppv_ratio:
139			random_state = RandomState(seed=bus_id)
140
141			n = max(int(len(pv_df) * grid_ratio), 1)
142
143			best_df = pv_df.sample(n=n, random_state=random_state)
144
145			i = 0
146
147			while (
148			not np.isclose(best_df.capacity.sum(), bat_cap, rtol=rtol)
149			and i < max_it
150			):
151			sample_df = pv_df.sample(n=n, random_state=random_state)
152
153			if abs(best_df.capacity.sum() - bat_cap) > abs(
154			sample_df.capacity.sum() - bat_cap
155			):
156			best_df = sample_df.copy()
157
158			i += 1
159
160			if sample_df.capacity.sum() < bat_cap:
161			n = min(n + 1, len(pv_df))
162			else:
163			n = max(n - 1, 1)
164
165			if not np.isclose(best_df.capacity.sum(), bat_cap, rtol=rtol):
166			logger.warning(
167			f"No suitable generators could be found in Grid "
168			f"{bus_id} and scenario {scenario} to achieve the "
169			f"desired ratio between battery capacity and pv "
170			f"rooftop capacity. The ratio will be "
171			f"{bat_cap / best_df.capacity.sum()}."
172			)
173
174			pv_df = best_df.copy()
175
176			bat_df = pv_df.drop(columns=["capacity"]).assign(
177			capacity=pv_df.capacity / pv_df.capacity.sum() * bat_cap,
178			p_nom=pv_df.capacity
179			/ pv_df.capacity.sum()
180			* bat_cap
181			/ cbat_pbat_ratio,
182			scenario=scenario,
183			bus_id=bus_id,
184			)
185
186			df_list.append(bat_df)
187
188			create_table(pd.concat(df_list, ignore_index=True))
189
190			add_metadata()
191
192
193			class EgonHomeBatteries(Base):
194			targets = config.datasets()["home_batteries"]["targets"]
195
196			__tablename__ = targets["home_batteries"]["table"]
197			__table_args__ = {"schema": targets["home_batteries"]["schema"]}
198
199			index = Column(Integer, primary_key=True, index=True)
200			scenario = Column(String)
201			bus_id = Column(Integer)
202			building_id = Column(Integer)
203			p_nom = Column(Float)
204			capacity = Column(Float)
205
206
207			def add_metadata():
208			"""
209			Add metadata to table supply.egon_home_batteries
210			"""
211			targets = config.datasets()["home_batteries"]["targets"]
212			deposit_id_mastr = config.datasets()["mastr_new"]["deposit_id"]
213			deposit_id_data_bundle = config.datasets()["data-bundle"]["sources"][
214			"zenodo"
215			]["deposit_id"]
216
217			contris = contributors(["kh", "kh"])
218
219			contris[0]["date"] = "2023-03-15"
220
221			contris[0]["object"] = "metadata"
222			contris[1]["object"] = "dataset"
223
224			contris[0]["comment"] = "Add metadata to dataset."
225			contris[1]["comment"] = "Add workflow to generate dataset."
226
227			meta = {
228			"name": (
229			f"{targets['home_batteries']['schema']}."
230			f"{targets['home_batteries']['table']}"
231			),
232			"title": "eGon Home Batteries",
233			"id": "WILL_BE_SET_AT_PUBLICATION",
234			"description": "Home storage systems allocated to buildings",
235			"language": "en-US",
236			"keywords": ["battery", "batteries", "home", "storage", "building"],
237			"publicationDate": datetime.date.today().isoformat(),
238			"context": context(),
239			"spatial": {
240			"location": "none",
241			"extent": "Germany",
242			"resolution": "building",
243			},
244			"temporal": {
245			"referenceDate": "2021-12-31",
246			"timeseries": {},
247			},
248			"sources": [
249			{
250			"title": "Data bundle for egon-data",
251			"description": (
252			"Data bundle for egon-data: A transparent and "
253			"reproducible data processing pipeline for energy "
254			"system modeling"
255			),
256			"path": (
257			"https://zenodo.org/record/"
258			f"{deposit_id_data_bundle}#.Y_dWM4CZMVM"
259			),
260			"licenses": [license_dedl(attribution="© Cußmann, Ilka")],
261			},
262			{
263			"title": ("open-MaStR power unit registry for eGo^n project"),
264			"description": (
265			"Data from Marktstammdatenregister (MaStR) data using "
266			"the data dump from 2022-11-17 for eGon-data."
267			),
268			"path": (
269			f"https://zenodo.org/record/{deposit_id_mastr}"
270			),
271			"licenses": [license_dedl(attribution="© Amme, Jonathan")],
272			},
273			sources()["openstreetmap"],
274			sources()["era5"],
275			sources()["vg250"],
276			sources()["egon-data"],
277			sources()["nep2021"],
278			sources()["mastr"],
279			sources()["technology-data"],
280			],
281			"licenses": [license_odbl("© eGon development team")],
282			"contributors": contris,
283			"resources": [
284			{
285			"profile": "tabular-data-resource",
286			"name": (
287			f"{targets['home_batteries']['schema']}."
288			f"{targets['home_batteries']['table']}"
289			),
290			"path": "None",
291			"format": "PostgreSQL",
292			"encoding": "UTF-8",
293			"schema": {
294			"fields": generate_resource_fields_from_db_table(
295			targets["home_batteries"]["schema"],
296			targets["home_batteries"]["table"],
297			),
298			"primaryKey": "index",
299			},
300			"dialect": {"delimiter": "", "decimalSeparator": ""},
301			}
302			],
303			"review": {"path": "", "badge": ""},
304			"metaMetadata": meta_metadata(),
305			"_comment": {
306			"metadata": (
307			"Metadata documentation and explanation (https://github.com/"
308			"OpenEnergyPlatform/oemetadata/blob/master/metadata/v141/"
309			"metadata_key_description.md)"
310			),
311			"dates": (
312			"Dates and time must follow the ISO8601 including time zone "
313			"(YYYY-MM-DD or YYYY-MM-DDThh:mm:ss±hh)"
314			),
315			"units": "Use a space between numbers and units (100 m)",
316			"languages": (
317			"Languages must follow the IETF (BCP47) format (en-GB, en-US, "
318			"de-DE)"
319			),
320			"licenses": (
321			"License name must follow the SPDX License List "
322			"(https://spdx.org/licenses/)"
323			),
324			"review": (
325			"Following the OEP Data Review (https://github.com/"
326			"OpenEnergyPlatform/data-preprocessing/wiki)"
327			),
328			"none": "If not applicable use (none)",
329			},
330			}
331
332			dialect = get_dialect(meta_metadata())()
333
334			meta = dialect.compile_and_render(dialect.parse(json.dumps(meta)))
335
336			db.submit_comment(
337			f"'{json.dumps(meta)}'",
338			targets["home_batteries"]["schema"],
339			targets["home_batteries"]["table"],
340			)
341
342
343			def create_table(df):
344			"""Create mapping table home battery <-> building id"""
345			engine = db.engine()
346
347			EgonHomeBatteries.__table__.drop(bind=engine, checkfirst=True)
348			EgonHomeBatteries.__table__.create(bind=engine, checkfirst=True)
349
350			df.reset_index().to_sql(
351			name=EgonHomeBatteries.__table__.name,
352			schema=EgonHomeBatteries.__table__.schema,
353			con=engine,
354			if_exists="append",
355			index=False,
356			)
357

openego / eGon-data

Push — dev ( 7cf077...0e9721 )

get_cbat_pbat_ratio() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like