14468425417

Committed 15 Apr 2025 11:32AM UTC coverage: 10.177%. Remained the same

Build # 14468425417

Build Type

push

github

Committed by

romainsacchi

Commit Message

Black reformating

Run Details

0 of 2 new or added lines in 2 files covered. (0.0%)

2 existing lines in 2 files now uncovered.

201 of 1975 relevant lines covered (10.18%)

0.1 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/carculator_utils/array.py

import itertools

import numpy as np
import pandas as pd
import xarray as xr

from .vehicle_input_parameters import VehicleInputParameters as vip


def fill_xarray_from_input_parameters(input_parameters, sensitivity=False, scope=None):
    """Create an `xarray` labeled array from the sampled input parameters.

    This function extracts the parameters' names and values contained in the
    `parameters` attribute of the :class:`CarInputParameters` class
    in :mod:`car_input_parameters` and insert them into a
    multi-dimensional numpy-like array from the *xarray* package
    (http://xarray.pydata.org/en/stable/).


    :param sensitivity:
    :param input_parameters: Instance of the :class:`TruckInputParameters` class
    in :mod:`truck_input_parameters`.
    :returns: `tuple`, `xarray.DataArray`
    - tuple (`size_dict`, `powertrain_dict`, `parameter_dict`, `year_dict`)
    - array

    Dimensions of `array`:

        0. Vehicle size, e.g. "3.5t", "7.5t", etc. str.
        1. Powertrain, e.g. "ICE-d", "BEV". str.
        2. Year. int.
        3. Samples.

    """

    # Check whether the argument passed is an instance of :class:`TruckInputParameters`
    if not isinstance(input_parameters, vip):
        raise TypeError(
            "The argument passed is not an object of the TruckInputParameter class"
        )

    if scope is None:
        scope = {
            "size": input_parameters.sizes,
            "powertrain": input_parameters.powertrains,
            "year": input_parameters.years,
        }
    else:
        if "size" not in scope:
            scope["size"] = input_parameters.sizes
        if "powertrain" not in scope:
            scope["powertrain"] = input_parameters.powertrains
        if "year" not in scope:
            scope["year"] = input_parameters.years

    # Make sure to include PHEV-e and PHEV-c-d if
    # PHEV-d is listed

    missing_pwts = [
        ("PHEV-d", "PHEV-e", "PHEV-c-d"),
        ("PHEV-p", "PHEV-e", "PHEV-c-p"),
    ]

    for missing_pwt in missing_pwts:
        if missing_pwt[0] in scope["powertrain"]:
            for p in missing_pwt[1:]:
                if not p in scope["powertrain"]:
                    scope["powertrain"].append(p)

    if any(s for s in scope["size"] if s not in input_parameters.sizes):
        raise ValueError("One of the size types is not valid.")

    if any(y for y in scope["year"] if y not in input_parameters.years):
        raise ValueError("One of the years defined is not valid.")

    if any(pt for pt in scope["powertrain"] if pt not in input_parameters.powertrains):
        raise ValueError("One of the powertrain types is not valid.")

    # if the purpose is not to do a sensitivity analysis
    # the dimension `value` of the array is as large as
    # the number of iterations to perform
    # that is, 1 in `static` mode, or several in `stochastic` mode.

    size_dict = {k: i for i, k in enumerate(scope["size"])}
    powertrain_dict = {k: i for i, k in enumerate(scope["powertrain"])}
    year_dict = {k: i for i, k in enumerate(scope["year"])}
    parameter_dict = {k: i for i, k in enumerate(input_parameters.parameters)}

    params = ["reference"] + input_parameters.input_parameters

    data_dict = [dict()]
    parameter_list = set()
    for param in input_parameters:
        pwt = (
            set(input_parameters.metadata[param]["powertrain"])
            if isinstance(input_parameters.metadata[param]["powertrain"], list)
            else set([input_parameters.metadata[param]["powertrain"]])
        )

        size = (
            set(input_parameters.metadata[param]["sizes"])
            if isinstance(input_parameters.metadata[param]["sizes"], list)
            else set([input_parameters.metadata[param]["sizes"]])
        )

        year = (
            set(input_parameters.metadata[param]["year"])
            if isinstance(input_parameters.metadata[param]["year"], list)
            else set([input_parameters.metadata[param]["year"]])
        )
        if (
            pwt.intersection(scope["powertrain"])
            and size.intersection(scope["size"])
            and year.intersection(scope["year"])
        ):
            powertrains = list(pwt.intersection(scope["powertrain"]))
            years = list(year.intersection(scope["year"]))
            sizes = list(size.intersection(scope["size"]))
            if len(sizes) > 1 and len(powertrains) > 1:
                pwt_size_couple = np.array(list(itertools.product(powertrains, sizes)))
                powertrains = pwt_size_couple[:, 0]
                sizes = pwt_size_couple[:, 1]

            data = {
                "size": sizes,
                "powertrain": powertrains,
                "parameter": input_parameters.metadata[param]["name"],
                "year": years,
                "data": input_parameters.values[param],
            }
            if not sensitivity:
                data["value"] = np.arange(input_parameters.iterations or 1)
            else:
                data["value"] = params

            data_dict.append(data)

            parameter_list.add(input_parameters.metadata[param]["name"])

    parameter_diff = parameter_list.symmetric_difference(input_parameters.parameters)

    for param in parameter_diff:
        data = {
            "size": scope["size"],
            "powertrain": scope["powertrain"],
            "parameter": param,
            "year": scope["year"],
            "data": 0.0,
        }
        if not sensitivity:
            data["value"] = np.arange(input_parameters.iterations or 1)
        else:
            data["value"] = params
        data_dict.append(data)

    df = pd.DataFrame.from_dict(data_dict)
    cols = ["powertrain", "size", "value", "year", "parameter"]
    df1 = pd.concat(
        [
            df[x]
            .explode()
            .to_frame()
            .assign(g=lambda x: x.groupby(level=0).cumcount())
            .set_index("g", append=True)
            for x in cols
        ],
        axis=1,
    )

    df = df.drop(cols, axis=1).join(df1.droplevel(1))
    df[cols] = df[cols].apply(lambda x: x.ffill())

    df = df.explode("data", ignore_index=False)
    df["value"] = df.groupby(["size", "powertrain", "parameter", "year"]).cumcount()

    df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)
    df.dropna(inplace=True)
    df = df[~df.index.duplicated(keep="first")]
    df = df.reset_index()
    if input_parameters.iterations:
        df = df[df["value"].isin(range(input_parameters.iterations))]

    # Ensure index is correctly set
    df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)

    # Create the DataArray
    array = xr.DataArray.from_series(df["data"])

    # Optional: cast types
    array = array.astype("float32")
    array.coords["year"] = array.coords["year"].astype("int")
    array = array.dropna("value", how="all")
    array = array.fillna(0.0)

    if sensitivity:
        # we increase each value by 10% for each params excepting reference one

        for param in params[1:]:
            array.loc[dict(parameter=param, value=param)] *= 1.1

    return (size_dict, powertrain_dict, parameter_dict, year_dict), array

1	import itertools	×
2
3	import numpy as np	×
4	import pandas as pd	×
5	import xarray as xr	×
6
7	from .vehicle_input_parameters import VehicleInputParameters as vip	×
8
9
10	def fill_xarray_from_input_parameters(input_parameters, sensitivity=False, scope=None):	×
11	"""Create an `xarray` labeled array from the sampled input parameters.
12
13	This function extracts the parameters' names and values contained in the
14	`parameters` attribute of the :class:`CarInputParameters` class
15	in :mod:`car_input_parameters` and insert them into a
16	multi-dimensional numpy-like array from the xarray package
17	(http://xarray.pydata.org/en/stable/).
18
19
20	:param sensitivity:
21	:param input_parameters: Instance of the :class:`TruckInputParameters` class
22	in :mod:`truck_input_parameters`.
23	:returns: `tuple`, `xarray.DataArray`
24	- tuple (`size_dict`, `powertrain_dict`, `parameter_dict`, `year_dict`)
25	- array
26
27	Dimensions of `array`:
28
29	0. Vehicle size, e.g. "3.5t", "7.5t", etc. str.
30	1. Powertrain, e.g. "ICE-d", "BEV". str.
31	2. Year. int.
32	3. Samples.
33
34	"""
35
36	# Check whether the argument passed is an instance of :class:`TruckInputParameters`
37	if not isinstance(input_parameters, vip):	×
38	raise TypeError(	×
39	"The argument passed is not an object of the TruckInputParameter class"
40	)
41
42	if scope is None:	×
43	scope = {	×
44	"size": input_parameters.sizes,
45	"powertrain": input_parameters.powertrains,
46	"year": input_parameters.years,
47	}
48	else:
49	if "size" not in scope:	×
50	scope["size"] = input_parameters.sizes	×
51	if "powertrain" not in scope:	×
52	scope["powertrain"] = input_parameters.powertrains	×
53	if "year" not in scope:	×
54	scope["year"] = input_parameters.years	×
55
56	# Make sure to include PHEV-e and PHEV-c-d if
57	# PHEV-d is listed
58
59	missing_pwts = [	×
60	("PHEV-d", "PHEV-e", "PHEV-c-d"),
61	("PHEV-p", "PHEV-e", "PHEV-c-p"),
62	]
63
64	for missing_pwt in missing_pwts:	×
65	if missing_pwt[0] in scope["powertrain"]:	×
66	for p in missing_pwt[1:]:	×
67	if not p in scope["powertrain"]:	×
68	scope["powertrain"].append(p)	×
69
70	if any(s for s in scope["size"] if s not in input_parameters.sizes):	×
71	raise ValueError("One of the size types is not valid.")	×
72
73	if any(y for y in scope["year"] if y not in input_parameters.years):	×
74	raise ValueError("One of the years defined is not valid.")	×
75
76	if any(pt for pt in scope["powertrain"] if pt not in input_parameters.powertrains):	×
77	raise ValueError("One of the powertrain types is not valid.")	×
78
79	# if the purpose is not to do a sensitivity analysis
80	# the dimension `value` of the array is as large as
81	# the number of iterations to perform
82	# that is, 1 in `static` mode, or several in `stochastic` mode.
83
84	size_dict = {k: i for i, k in enumerate(scope["size"])}	×
85	powertrain_dict = {k: i for i, k in enumerate(scope["powertrain"])}	×
86	year_dict = {k: i for i, k in enumerate(scope["year"])}	×
87	parameter_dict = {k: i for i, k in enumerate(input_parameters.parameters)}	×
88
89	params = ["reference"] + input_parameters.input_parameters	×
90
91	data_dict = [dict()]	×
92	parameter_list = set()	×
93	for param in input_parameters:	×
94	pwt = (	×
95	set(input_parameters.metadata[param]["powertrain"])
96	if isinstance(input_parameters.metadata[param]["powertrain"], list)
97	else set([input_parameters.metadata[param]["powertrain"]])
98	)
99
100	size = (	×
101	set(input_parameters.metadata[param]["sizes"])
102	if isinstance(input_parameters.metadata[param]["sizes"], list)
103	else set([input_parameters.metadata[param]["sizes"]])
104	)
105
106	year = (	×
107	set(input_parameters.metadata[param]["year"])
108	if isinstance(input_parameters.metadata[param]["year"], list)
109	else set([input_parameters.metadata[param]["year"]])
110	)
111	if (	×
112	pwt.intersection(scope["powertrain"])
113	and size.intersection(scope["size"])
114	and year.intersection(scope["year"])
115	):
116	powertrains = list(pwt.intersection(scope["powertrain"]))	×
117	years = list(year.intersection(scope["year"]))	×
118	sizes = list(size.intersection(scope["size"]))	×
119	if len(sizes) > 1 and len(powertrains) > 1:	×
120	pwt_size_couple = np.array(list(itertools.product(powertrains, sizes)))	×
121	powertrains = pwt_size_couple[:, 0]	×
122	sizes = pwt_size_couple[:, 1]	×
123
124	data = {	×
125	"size": sizes,
126	"powertrain": powertrains,
127	"parameter": input_parameters.metadata[param]["name"],
128	"year": years,
129	"data": input_parameters.values[param],
130	}
131	if not sensitivity:	×
132	data["value"] = np.arange(input_parameters.iterations or 1)	×
133	else:
134	data["value"] = params	×
135
136	data_dict.append(data)	×
137
138	parameter_list.add(input_parameters.metadata[param]["name"])	×
139
140	parameter_diff = parameter_list.symmetric_difference(input_parameters.parameters)	×
141
142	for param in parameter_diff:	×
143	data = {	×
144	"size": scope["size"],
145	"powertrain": scope["powertrain"],
146	"parameter": param,
147	"year": scope["year"],
148	"data": 0.0,
149	}
150	if not sensitivity:	×
151	data["value"] = np.arange(input_parameters.iterations or 1)	×
152	else:
153	data["value"] = params	×
154	data_dict.append(data)	×
155
156	df = pd.DataFrame.from_dict(data_dict)	×
157	cols = ["powertrain", "size", "value", "year", "parameter"]	×
158	df1 = pd.concat(	×
159	[
160	df[x]
161	.explode()
162	.to_frame()
163	.assign(g=lambda x: x.groupby(level=0).cumcount())
164	.set_index("g", append=True)
165	for x in cols
166	],
167	axis=1,
168	)
169
UNCOV 170	df = df.drop(cols, axis=1).join(df1.droplevel(1))	×
171	df[cols] = df[cols].apply(lambda x: x.ffill())	×
172
173	df = df.explode("data", ignore_index=False)	×
174	df["value"] = df.groupby(["size", "powertrain", "parameter", "year"]).cumcount()	×
175
176	df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)	×
177	df.dropna(inplace=True)	×
178	df = df[~df.index.duplicated(keep="first")]	×
179	df = df.reset_index()	×
180	if input_parameters.iterations:	×
181	df = df[df["value"].isin(range(input_parameters.iterations))]	×
182
183	# Ensure index is correctly set
184	df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)	×
185
186	# Create the DataArray
187	array = xr.DataArray.from_series(df["data"])	×
188
189	# Optional: cast types
190	array = array.astype("float32")	×
191	array.coords["year"] = array.coords["year"].astype("int")	×
192	array = array.dropna("value", how="all")	×
193	array = array.fillna(0.0)	×
194
195	if sensitivity:	×
196	# we increase each value by 10% for each params excepting reference one
197
198	for param in params[1:]:	×
199	array.loc[dict(parameter=param, value=param)] *= 1.1	×
200
201	return (size_dict, powertrain_dict, parameter_dict, year_dict), array	×

romainsacchi / carculator_utils / 14468425417

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous