9514243606

Committed 14 Jun 2024 09:57AM UTC coverage: 10.292%. Remained the same

Build Type

push

github

Committed by romainsacchi

Commit Message

Merge remote-tracking branch 'origin/master'

Run Details

0 of 36 new or added lines in 1 file covered. (0.0%)

18 existing lines in 1 file now uncovered.

201 of 1953 relevant lines covered (10.29%)

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"]:
            if not any(p in scope["powertrain"] for p in missing_pwt[1:]):
                scope["powertrain"].extend(missing_pwt[1:])

    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.fillna(method="ffill"))
    df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)
    df.dropna(inplace=True)
    array = xr.DataArray.from_series(df["data"])
    array = array.astype("float32")
    array.coords["year"] = array.coords["year"].astype(int)
    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
UNCOV 7	from .vehicle_input_parameters import VehicleInputParameters as vip	×
8
9
UNCOV 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`
UNCOV 37	if not isinstance(input_parameters, vip):	×
UNCOV 38	raise TypeError(	×
39	"The argument passed is not an object of the TruckInputParameter class"
40	)
41
UNCOV 42	if scope is None:	×
UNCOV 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	×
UNCOV 53	if "year" not in scope:	×
UNCOV 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
UNCOV 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"]:	×
UNCOV 66	if not any(p in scope["powertrain"] for p in missing_pwt[1:]):	×
67	scope["powertrain"].extend(missing_pwt[1:])	×
68
UNCOV 69	if any(s for s in scope["size"] if s not in input_parameters.sizes):	×
70	raise ValueError("One of the size types is not valid.")	×
71
UNCOV 72	if any(y for y in scope["year"] if y not in input_parameters.years):	×
73	raise ValueError("One of the years defined is not valid.")	×
74
UNCOV 75	if any(pt for pt in scope["powertrain"] if pt not in input_parameters.powertrains):	×
UNCOV 76	raise ValueError("One of the powertrain types is not valid.")	×
77
78	# if the purpose is not to do a sensitivity analysis
79	# the dimension `value` of the array is as large as
80	# the number of iterations to perform
81	# that is, 1 in `static` mode, or several in `stochastic` mode.
82
83	size_dict = {k: i for i, k in enumerate(scope["size"])}	×
84	powertrain_dict = {k: i for i, k in enumerate(scope["powertrain"])}	×
UNCOV 85	year_dict = {k: i for i, k in enumerate(scope["year"])}	×
86	parameter_dict = {k: i for i, k in enumerate(input_parameters.parameters)}	×
87
NEW 88	params = ["reference"] + input_parameters.input_parameters	×
89
NEW 90	data_dict = [dict()]	×
NEW 91	parameter_list = set()	×
UNCOV 92	for param in input_parameters:	×
UNCOV 93	pwt = (	×
94	set(input_parameters.metadata[param]["powertrain"])
95	if isinstance(input_parameters.metadata[param]["powertrain"], list)
96	else set([input_parameters.metadata[param]["powertrain"]])
97	)
98
UNCOV 99	size = (	×
100	set(input_parameters.metadata[param]["sizes"])
101	if isinstance(input_parameters.metadata[param]["sizes"], list)
102	else set([input_parameters.metadata[param]["sizes"]])
103	)
104
105	year = (	×
106	set(input_parameters.metadata[param]["year"])
107	if isinstance(input_parameters.metadata[param]["year"], list)
108	else set([input_parameters.metadata[param]["year"]])
109	)
110	if (	×
111	pwt.intersection(scope["powertrain"])
112	and size.intersection(scope["size"])
113	and year.intersection(scope["year"])
114	):
115
NEW 116	powertrains = list(pwt.intersection(scope["powertrain"]))	×
NEW 117	years = list(year.intersection(scope["year"]))	×
NEW 118	sizes = list(size.intersection(scope["size"]))	×
NEW 119	if len(sizes) > 1 and len(powertrains) > 1:	×
NEW 120	pwt_size_couple = np.array(list(itertools.product(powertrains, sizes)))	×
NEW 121	powertrains = pwt_size_couple[:, 0]	×
NEW 122	sizes = pwt_size_couple[:, 1]	×
123
NEW 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	}
NEW 131	if not sensitivity:	×
NEW 132	data["value"] = np.arange(input_parameters.iterations or 1)	×
133	else:
NEW 134	data["value"] = params	×
135
NEW 136	data_dict.append(data)	×
137
NEW 138	parameter_list.add(input_parameters.metadata[param]["name"])	×
139
NEW 140	parameter_diff = parameter_list.symmetric_difference(input_parameters.parameters)	×
141
NEW 142	for param in parameter_diff:	×
NEW 143	data = {	×
144	"size": scope["size"],
145	"powertrain": scope["powertrain"],
146	"parameter": param,
147	"year": scope["year"],
148	"data": 0.0,
149	}
NEW 150	if not sensitivity:	×
NEW 151	data["value"] = np.arange(input_parameters.iterations or 1)	×
152	else:
NEW 153	data["value"] = params	×
NEW 154	data_dict.append(data)	×
155
NEW 156	df = pd.DataFrame.from_dict(data_dict)	×
NEW 157	cols = ["powertrain", "size", "value", "year", "parameter"]	×
NEW 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
NEW 170	df = df.drop(cols, axis=1).join(df1.droplevel(1))	×
NEW 171	df[cols] = df[cols].apply(lambda x: x.fillna(method="ffill"))	×
NEW 172	df.set_index(["size", "powertrain", "parameter", "year", "value"], inplace=True)	×
NEW 173	df.dropna(inplace=True)	×
NEW 174	array = xr.DataArray.from_series(df["data"])	×
NEW 175	array = array.astype("float32")	×
NEW 176	array.coords["year"] = array.coords["year"].astype(int)	×
NEW 177	array = array.fillna(0.0)	×
178
NEW 179	if sensitivity:	×
180	# we increase each value by 10% for each params excepting reference one
181
NEW 182	for param in params[1:]:	×
183	array.loc[dict(parameter=param, value=param)] *= 1.1	×
184
185	return (size_dict, powertrain_dict, parameter_dict, year_dict), array	×

romainsacchi / carculator_utils / 9514243606

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