import logging
import os
import pickle
import pprint as pp
import warnings
from deflex.scenario import DeflexScenario
[docs]def search_dumped_scenarios(path, extension="dflx", **parameter_filter):
"""Filter results by extension and meta data.
The function will search the $HOME folder recursively for files with the
'.dflx' extension. Afterwards all files will filtered by the meta data.
If there is an `info` table in your input data, the keys and values can be
used to filter the values. For example different region sets are defined
as maps with `de21`, `de22` and `de17` and different years were modelled.
Futhermore some are modelled with the heating sector (heat: True) and some
not (heat: False). See the example below on how to search for these
scenarios.
Parameters
----------
path : str
Start folder from where to search recursively.
extension : str
Extension of the results files (default: ".dflx")
**parameter_filter
Set filter always with lists e.g. map=["de21"] or map=["de21", "de22"].
The values in the list have to be strings. Two filters will be
connected with 'AND', the values within one filter with `OR`.
The filters year=["2014"], map=["de21", "de22"] will find all scenarios
with: year==2014 and (map=="de21" or map=="de22")
Returns
-------
Examples
--------
>>> from deflex import TEST_PATH
>>> from deflex import fetch_test_files
>>> my_file_name = fetch_test_files("de17_heat.dflx")
>>> res = search_dumped_scenarios(path=TEST_PATH, map=["de17"])
>>> len(res)
2
>>> sorted(res)[0].split(os.sep)[-1]
'de17_heat.dflx'
>>> res = search_dumped_scenarios(path=TEST_PATH, map=["de17", "de21"])
>>> len(res)
6
>>> res = search_dumped_scenarios(
... path=TEST_PATH, map=["de17", "de21"], heat=["True"])
>>> len(res)
3
>>> sorted(res)[0].split(os.sep)[-1]
'de17_heat.dflx'
"""
result_files = []
for root, dirs, files in os.walk(path):
files = [f for f in files if not f[0] == "."]
dirs[:] = [d for d in dirs if not d[0] == "."]
if "." + extension in str(files):
for f in files:
if f.split(".")[-1] == extension:
result_files.append(os.path.join(root, f))
files = {}
# filter by meta data.
for name in result_files:
fn = os.path.join(path, name)
f = open(fn, "rb")
files[name] = pickle.load(f)
f.close()
for filter_key, filter_value in parameter_filter.items():
files = {
k: v
for k, v in files.items()
if any(
[
str(v.get(filter_key)).lower() == str(f).lower()
for f in filter_value
]
)
}
return list(files.keys())
def search_results(path, extension="dflx", **parameter_filter):
"""Keep the old name to keep the old API"""
msg = (
"'search_results' is deprecated. Use 'search_dumped_scenarios` "
"instead."
)
warnings.warn(msg, FutureWarning)
search_dumped_scenarios(path, extension, **parameter_filter)
[docs]def restore_results(file_names, scenario_class=DeflexScenario):
"""
Restore only the result dictionary from a dumped scenario or a list of
dumped scenarios. The results will be a deflex result dictionary with the
following keys:
* main – Results of all variables
* param – Input parameter
* meta – Meta information and tags of the scenario
* problem – Information about the linear problem such as lower bound,
upper bound etc.
* solver – Solver results
* solution – Information about the found solution and the objective value
Parameters
----------
file_names : list or string
All file names (full path) that should be loaded.
scenario_class : class
A child of the deflex.Scenario class or the Scenario class itself.
Returns
-------
A list of results dictionaries or a single dictionary if one file name is
given : list or dict
Examples
--------
>>> from deflex import fetch_test_files
>>> fn1 = fetch_test_files("de21_no-heat_transmission.dflx")
>>> fn2 = fetch_test_files("de02_no-heat.dflx")
>>> sorted(restore_results(fn1).keys())
['Input data', 'Main', 'Meta', 'Param', 'Problem', 'Solution', 'Solver']
>>> sorted(restore_results([fn1, fn2])[0].keys())
['Input data', 'Main', 'Meta', 'Param', 'Problem', 'Solution', 'Solver']
"""
if not isinstance(file_names, list):
file_names = list((file_names,))
results = []
for path in file_names:
sc = restore_scenario(path, scenario_class)
tmp_res = sc.results
tmp_res["meta"]["filename"] = os.path.basename(path)
tmp_res["input_data"] = sc.input_data
results.append(tmp_res)
if len(results) < 2:
results = results[0]
return results
[docs]def restore_scenario(filename, scenario_class=DeflexScenario):
"""
Restore a full Scenario from a dump file (`.dflx`).
If only the results are needed use
:py:func:`~deflex.restore_results` instead.
By default a DeflexScenario is created but a different Scenario class can
be passed The Scenario has to be equal to the dumped Scenario otherwise the
restore will fail.
Parameters
----------
filename : str
The path to the dumped file (`.dflx`).
scenario_class : class
A child of the deflex.Scenario class or the Scenario class itself.
Returns
-------
deflex.Scenario
"""
if filename.split(".")[-1] != "dflx":
msg = (
"The suffix of a valid deflex scenario has to be '.dflx'.\n"
"Cannot open {0}.".format(filename)
)
raise IOError(msg)
f = open(filename, "rb")
meta = pickle.load(f)
logging.info("Meta information:\n %s", pp.pformat(meta))
sc = scenario_class()
sc.__dict__ = pickle.load(f)
f.close()
logging.info("Results restored from %s.", filename)
return sc
[docs]def create_scenario(path, file_type=None):
"""
Create a deflex scenario object from file.
Parameters
----------
path : str
A valid deflex scenario file.
file_type : str or None
Type of the input data. Valid values are 'csv', 'xlsx', None. If the
input is non the path should end on 'csv', '.xlsx' to allow
auto-detection.
Returns
-------
deflex.DeflexScenario
Examples
--------
>>> from deflex import fetch_test_files, TEST_PATH
>>> fn = fetch_test_files("de17_heat.xlsx")
>>> s = create_scenario(fn, file_type="xlsx")
>>> type(s)
<class 'deflex.scenario.DeflexScenario'>
>>> int(s.input_data["volatile plants"]["capacity"]["DE01", "wind"])
3815
>>> type(create_scenario(fn))
<class 'deflex.scenario.DeflexScenario'>
>>> create_scenario(fn, file_type="csv"
... ) # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
...
NotADirectoryError: [Errno 20] Not a directory:
"""
sc = DeflexScenario()
if path is not None:
if file_type is None:
if ".xlsx" in path[-5:]:
file_type = "xlsx"
elif "csv" in path[-4:]:
file_type = "csv"
else:
file_type = None
logging.info("Reading file: %s", path)
if file_type == "xlsx":
sc.read_xlsx(path)
elif file_type == "csv":
sc.read_csv(path)
return sc