13394719474

Committed 18 Feb 2025 03:55PM UTC coverage: 96.919% (-1.0%) from 97.93%

Build # 13394719474

Build Type

push

github

Committed by

sebastian-swob

Commit Message

increased positional arguments to 7

Run Details

1164 of 1201 relevant lines covered (96.92%)

1.93 hits per line

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

81.72

/pytrnsys_process/utils.py

import collections.abc as _abc
import logging as _logging
import os as _os
import pathlib as _pl
import pickle as _pickle
import subprocess as _subprocess
from typing import cast

import matplotlib.pyplot as _plt

from pytrnsys_process import data_structures as ds
from pytrnsys_process import logger as log
from pytrnsys_process import settings as sett


def get_sim_folders(path_to_results: _pl.Path) -> _abc.Sequence[_pl.Path]:
    sim_folders = []
    for item in path_to_results.glob("*"):
        if item.is_dir():
            sim_folders.append(item)
    return sim_folders


def get_files(
        sim_folders: _abc.Sequence[_pl.Path],
        results_folder_name: str = sett.settings.reader.folder_name_for_printer_files,
        get_mfr_and_t: bool = sett.settings.reader.read_step_files,
        read_deck_files: bool = sett.settings.reader.read_deck_files,
) -> _abc.Sequence[_pl.Path]:
    """Get simulation files from folders based on configuration.

    Args:
        sim_folders: Sequence of paths to simulation folders
        results_folder_name: Name of folder containing printer files
        get_mfr_and_t: Whether to include step files (T and Mfr files)
        read_deck_files: Whether to include deck files

    Returns:
        Sequence of paths to simulation files
    """
    sim_files: list[_pl.Path] = []
    for sim_folder in sim_folders:
        if get_mfr_and_t:
            sim_files.extend(sim_folder.glob("*[_T,_Mfr].prt"))
        if read_deck_files:
            sim_files.extend(sim_folder.glob("**/*.dck"))
        results_path = sim_folder / results_folder_name
        if results_path.exists():
            sim_files.extend(results_path.glob("*"))

    return [x for x in sim_files if x.is_file()]


# TODO add docstring #pylint: disable=fixme


def export_plots_in_configured_formats(
        fig: _plt.Figure,
        path_to_directory: _pl.Path,
        plot_name: str,
        plots_folder_name: str = "plots",
) -> None:
    """Save a matplotlib figure in multiple formats and sizes.

        Saves the figure in all configured formats (png, pdf, emf) and sizes (A4, A4_HALF)
        as specified in the plot settings (api.settings.plot).
        For EMF format, the figure is first saved as SVG and then converted using Inkscape.

        Args:
            fig: The matplotlib Figure object to save.
            path_to_directory: Directory path where the plots should be saved.
            plot_name: Base name for the plot file (will be appended with size and format).
            plots_folder_name: leave empty if you don't want to save in a new folder

        Returns:
            None

        Note:
            - Creates a 'plots' subdirectory if it doesn't exist
            - For EMF files, requires Inkscape to be installed at the configured path
            - File naming format: {plot_name}-{size_name}.{format}

        Example:
    import data_structures    >>> from pytrnsys_process import api
        >>> def processing_of_monthly_data(simulation: data_structures.Simulation):
        >>>     monthly_df = simulation.monthly
        >>>     columns_to_plot = ["QSnk60P", "QSnk60PauxCondSwitch_kW"]
        >>>     fig, ax = api.bar_chart(monthly_df, columns_to_plot)
        >>>
        >>>     # Save the plot in multiple formats
        >>>     api.export_plots_in_configured_formats(fig, simulation.path, "monthly-bar-chart")
        >>>     # Creates files like:
        >>>     #   results/simulation1/plots/monthly-bar-chart-A4.png
        >>>     #   results/simulation1/plots/monthly-bar-chart-A4.pdf
        >>>     #   results/simulation1/plots/monthly-bar-chart-A4.emf
        >>>     #   results/simulation1/plots/monthly-bar-chart-A4_HALF.png
        >>>     #   etc.

    """
    plot_settings = sett.settings.plot
    plots_folder = path_to_directory / plots_folder_name
    plots_folder.mkdir(exist_ok=True)

    for size_name, size in plot_settings.figure_sizes.items():
        file_no_suffix = plots_folder / f"{plot_name}-{size_name}"
        fig.set_size_inches(size)
        for fmt in plot_settings.file_formats:
            if fmt == ".emf":
                path_to_svg = file_no_suffix.with_suffix(".svg")
                fig.savefig(path_to_svg)
                convert_svg_to_emf(file_no_suffix)
                if ".svg" not in plot_settings.file_formats:
                    _os.remove(path_to_svg)
            else:
                fig.savefig(file_no_suffix.with_suffix(fmt))


def convert_svg_to_emf(file_no_suffix: _pl.Path) -> None:
    try:
        inkscape_path = sett.settings.plot.inkscape_path
        if not _pl.Path(inkscape_path).exists():
            raise OSError(f"Inkscape executable not found at: {inkscape_path}")
        emf_filepath = file_no_suffix.with_suffix(".emf")
        path_to_svg = file_no_suffix.with_suffix(".svg")

        _subprocess.run(
            [
                inkscape_path,
                "--export-filename=" + str(emf_filepath),
                "--export-type=emf",
                str(path_to_svg),
            ],
            check=True,
            capture_output=True,
            text=True,
        )

    except _subprocess.CalledProcessError as e:
        log.main_logger.error(
            "Inkscape conversion failed: %s\nOutput: %s",
            e,
            e.output,
            exc_info=True,
        )
    except OSError as e:
        log.main_logger.error(
            "System error running Inkscape: %s", e, exc_info=True
        )


def get_file_content_as_string(
        file_path: _pl.Path, encoding: str = "UTF-8"
) -> str:
    """Read and return the entire content of a file as a string.

    Args:
        file_path (Path): Path to the file to read
        encoding (str, optional): File encoding to use. Defaults to "UTF-8".

    Returns:
        str: Content of the file as a string
    """
    with open(file_path, "r", encoding=encoding) as file:
        return file.read()


# def save_results_for_comparison(results: ResultsForComparison, path: _pl.Path) -> None:
#     """Save ResultsForComparison data to a JSON file.
#
#     This function saves monthly, hourly, step and scalar data from a ResultsForComparison
#     object to a single JSON file.
#
#     Args:
#         results: ResultsForComparison object to save
#         path: Path where to save the JSON file
#
#     Returns:
#         None
#
#     Raises:
#         OSError: If there's an error writing to the file
#     """
#     data = {
#         'path_to_simulations': str(results.path_to_simulations),
#         'monthly': {
#             name: {
#                 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
#                 'columns': df.columns.tolist() if not df.empty else [],
#                 'data': df.values.tolist() if not df.empty else []
#             }
#             for name, df in results.monthly.items()
#         },
#         'hourly': {
#             name: {
#                 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
#                 'columns': df.columns.tolist() if not df.empty else [],
#                 'data': df.values.tolist() if not df.empty else []
#             }
#             for name, df in results.hourly.items()
#         },
#         'step': {
#             name: {
#                 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
#                 'columns': df.columns.tolist() if not df.empty else [],
#                 'data': df.values.tolist() if not df.empty else []
#             }
#             for name, df in results.step.items()
#         },
#         'scalar': {
#             'index': results.scalar.index.tolist() if not results.scalar.empty else [],
#             'columns': results.scalar.columns.tolist() if not results.scalar.empty else [],
#             'data': results.scalar.values.tolist() if not results.scalar.empty else []
#         }
#     }
#
#     try:
#         with open(path, 'w', encoding='utf-8') as f:
#             _json.dump(data, f, indent=2)
#     except OSError as e:
#         logger.error("Error saving ResultsForComparison to JSON: %s", e, exc_info=True)
#         raise
#
# def load_results_for_comparison(path: _pl.Path) -> ResultsForComparison:
#     """Load ResultsForComparison data from a JSON file.
#
#     This function loads monthly, hourly, step and scalar data from a JSON file
#     and reconstructs a ResultsForComparison object.
#
#     Args:
#         path: Path to the JSON file to load
#
#     Returns:
#         ResultsForComparison: Reconstructed ResultsForComparison object
#
#     Raises:
#         OSError: If there's an error reading the file
#         ValueError: If the file format is invalid
#     """
#     try:
#         with open(path, 'r', encoding='utf-8') as f:
#             data = _json.load(f)
#
#         results = ResultsForComparison()
#         results.path_to_simulations = _pl.Path(data['path_to_simulations'])
#
#         # Load monthly data
#         results.monthly = {
#             name: _pd.DataFrame(
#                 data=df_data.get('data', []),
#                 index=_pd.to_datetime(df_data.get('index', [])),
#                 columns=df_data.get('columns', [])
#             )
#             for name, df_data in data.get('monthly', {}).items()
#         }
#
#         # Load hourly data
#         results.hourly = {
#             name: _pd.DataFrame(
#                 data=df_data.get('data', []),
#                 index=_pd.to_datetime(df_data.get('index', [])),
#                 columns=df_data.get('columns', [])
#             )
#             for name, df_data in data.get('hourly', {}).items()
#         }
#
#         # Load step data
#         results.step = {
#             name: _pd.DataFrame(
#                 data=df_data.get('data', []),
#                 index=_pd.to_datetime(df_data.get('index', [])),
#                 columns=df_data.get('columns', [])
#             )
#             for name, df_data in data.get('step', {}).items()
#         }
#
#         # Load scalar data
#         scalar_data = data.get('scalar', {})
#         results.scalar = _pd.DataFrame(
#             data=scalar_data.get('data', []),
#             index=scalar_data.get('index', []),
#             columns=scalar_data.get('columns', [])
#         )
#
#         return results
#
#     except OSError as e:
#         logger.error("Error loading ResultsForComparison from JSON: %s", e, exc_info=True)
#         raise
#     except (KeyError, ValueError, _json.JSONDecodeError) as e:
#         logger.error("Invalid ResultsForComparison JSON format: %s", e, exc_info=True)
#         raise ValueError(f"Invalid ResultsForComparison JSON format: {str(e)}")


def save_to_pickle(
        data: ds.Simulation | ds.SimulationsData,
        path: _pl.Path,
        logger: _logging.Logger = log.main_logger,
) -> None:
    """Save ResultsForComparison data to a pickle file.

    This function saves the entire ResultsForComparison object to a pickle file,
    preserving all data structures and relationships.

    Args:
        data: data object to save
        path: Path where to save the pickle file

    Returns:
        None

    Raises:
        OSError: If there's an error writing to the file
    """
    try:
        with open(path, "wb") as f:
            _pickle.dump(data, f)
    except OSError as e:
        logger.error(
            "Error saving ResultsForComparison to pickle: %s", e, exc_info=True
        )
        raise


def load_simulations_data_from_pickle(
        path: _pl.Path, logger: _logging.Logger = log.main_logger
) -> ds.SimulationsData:
    """Load ResultsForComparison data from a pickle file.

    This function loads a previously saved ResultsForComparison object from a pickle file.

    Args:
        path: Path to the pickle file to load

    Returns:
        ResultsForComparison: Reconstructed ResultsForComparison object

    Raises:
        OSError: If there's an error reading the file
        pickle.UnpicklingError: If the file is corrupted or invalid
    """
    try:
        with open(path, "rb") as f:
            simulations_data = _pickle.load(f)

        # Check if it has the expected attributes of SimulationsData
        required_attrs = {'simulations', 'scalar', 'path_to_simulations'}
        if all(hasattr(simulations_data, attr) for attr in required_attrs):
            return cast(ds.SimulationsData, simulations_data)

        raise ValueError(
            f"Loaded object is missing required SimulationsData attributes. Type: {type(simulations_data).__name__}"
        )

    except OSError as e:
        logger.error(
            "Error loading ResultsForComparison from pickle: %s",
            e,
            exc_info=True,
        )
        raise
    except (_pickle.UnpicklingError, ValueError) as e:
        logger.error(
            "Invalid ResultsForComparison pickle format: %s", e, exc_info=True
        )
        raise


def load_simulation_from_pickle(
        path: _pl.Path, logger: _logging.Logger = log.main_logger
) -> ds.Simulation:
    try:
        with open(path, "rb") as f:
            simulation = _pickle.load(f)

        # Check if it has the expected attributes of a Simulation
        required_attrs = {'monthly', 'hourly', 'step', 'scalar', 'path'}
        if all(hasattr(simulation, attr) for attr in required_attrs):
            return cast(ds.Simulation, simulation)

        raise ValueError(
            f"Loaded object is missing required Simulation attributes. Type: {type(simulation).__name__}"
        )
    except OSError as e:
        logger.error(
            "Error loading Simulation from pickle: %s", e, exc_info=True
        )
        raise
    except (_pickle.UnpicklingError, ValueError) as e:
        logger.error("Invalid Simulation pickle format: %s", e, exc_info=True)
        raise

1	import collections.abc as _abc	2✔
2	import logging as _logging	2✔
3	import os as _os	2✔
4	import pathlib as _pl	2✔
5	import pickle as _pickle	2✔
6	import subprocess as _subprocess	2✔
7	from typing import cast	2✔
8
9	import matplotlib.pyplot as _plt	2✔
10
11	from pytrnsys_process import data_structures as ds	2✔
12	from pytrnsys_process import logger as log	2✔
13	from pytrnsys_process import settings as sett	2✔
14
15
16	def get_sim_folders(path_to_results: _pl.Path) -> _abc.Sequence[_pl.Path]:	2✔
17	sim_folders = []	2✔
18	for item in path_to_results.glob("*"):	2✔
19	if item.is_dir():	2✔
20	sim_folders.append(item)	2✔
21	return sim_folders	2✔
22
23
24	def get_files(	2✔
25	sim_folders: _abc.Sequence[_pl.Path],
26	results_folder_name: str = sett.settings.reader.folder_name_for_printer_files,
27	get_mfr_and_t: bool = sett.settings.reader.read_step_files,
28	read_deck_files: bool = sett.settings.reader.read_deck_files,
29	) -> _abc.Sequence[_pl.Path]:
30	"""Get simulation files from folders based on configuration.
31
32	Args:
33	sim_folders: Sequence of paths to simulation folders
34	results_folder_name: Name of folder containing printer files
35	get_mfr_and_t: Whether to include step files (T and Mfr files)
36	read_deck_files: Whether to include deck files
37
38	Returns:
39	Sequence of paths to simulation files
40	"""
41	sim_files: list[_pl.Path] = []	2✔
42	for sim_folder in sim_folders:	2✔
43	if get_mfr_and_t:	2✔
44	sim_files.extend(sim_folder.glob("*[_T,_Mfr].prt"))	2✔
45	if read_deck_files:	2✔
46	sim_files.extend(sim_folder.glob("*/.dck"))	2✔
47	results_path = sim_folder / results_folder_name	2✔
48	if results_path.exists():	2✔
49	sim_files.extend(results_path.glob("*"))	2✔
50
51	return [x for x in sim_files if x.is_file()]	2✔
52
53
54	# TODO add docstring #pylint: disable=fixme
55
56
57	def export_plots_in_configured_formats(	2✔
58	fig: _plt.Figure,
59	path_to_directory: _pl.Path,
60	plot_name: str,
61	plots_folder_name: str = "plots",
62	) -> None:
63	"""Save a matplotlib figure in multiple formats and sizes.
64
65	Saves the figure in all configured formats (png, pdf, emf) and sizes (A4, A4_HALF)
66	as specified in the plot settings (api.settings.plot).
67	For EMF format, the figure is first saved as SVG and then converted using Inkscape.
68
69	Args:
70	fig: The matplotlib Figure object to save.
71	path_to_directory: Directory path where the plots should be saved.
72	plot_name: Base name for the plot file (will be appended with size and format).
73	plots_folder_name: leave empty if you don't want to save in a new folder
74
75	Returns:
76	None
77
78	Note:
79	- Creates a 'plots' subdirectory if it doesn't exist
80	- For EMF files, requires Inkscape to be installed at the configured path
81	- File naming format: {plot_name}-{size_name}.{format}
82
83	Example:
84	import data_structures >>> from pytrnsys_process import api
85	>>> def processing_of_monthly_data(simulation: data_structures.Simulation):
86	>>> monthly_df = simulation.monthly
87	>>> columns_to_plot = ["QSnk60P", "QSnk60PauxCondSwitch_kW"]
88	>>> fig, ax = api.bar_chart(monthly_df, columns_to_plot)
89	>>>
90	>>> # Save the plot in multiple formats
91	>>> api.export_plots_in_configured_formats(fig, simulation.path, "monthly-bar-chart")
92	>>> # Creates files like:
93	>>> # results/simulation1/plots/monthly-bar-chart-A4.png
94	>>> # results/simulation1/plots/monthly-bar-chart-A4.pdf
95	>>> # results/simulation1/plots/monthly-bar-chart-A4.emf
96	>>> # results/simulation1/plots/monthly-bar-chart-A4_HALF.png
97	>>> # etc.
98
99	"""
100	plot_settings = sett.settings.plot	2✔
101	plots_folder = path_to_directory / plots_folder_name	2✔
102	plots_folder.mkdir(exist_ok=True)	2✔
103
104	for size_name, size in plot_settings.figure_sizes.items():	2✔
105	file_no_suffix = plots_folder / f"{plot_name}-{size_name}"	2✔
106	fig.set_size_inches(size)	2✔
107	for fmt in plot_settings.file_formats:	2✔
108	if fmt == ".emf":	2✔
109	path_to_svg = file_no_suffix.with_suffix(".svg")	2✔
110	fig.savefig(path_to_svg)	2✔
111	convert_svg_to_emf(file_no_suffix)	2✔
112	if ".svg" not in plot_settings.file_formats:	2✔
113	_os.remove(path_to_svg)	2✔
114	else:
115	fig.savefig(file_no_suffix.with_suffix(fmt))	2✔
116
117
118	def convert_svg_to_emf(file_no_suffix: _pl.Path) -> None:	2✔
119	try:	2✔
120	inkscape_path = sett.settings.plot.inkscape_path	2✔
121	if not _pl.Path(inkscape_path).exists():	2✔
122	raise OSError(f"Inkscape executable not found at: {inkscape_path}")	2✔
123	emf_filepath = file_no_suffix.with_suffix(".emf")	2✔
124	path_to_svg = file_no_suffix.with_suffix(".svg")	2✔
125
126	_subprocess.run(	2✔
127	[
128	inkscape_path,
129	"--export-filename=" + str(emf_filepath),
130	"--export-type=emf",
131	str(path_to_svg),
132	],
133	check=True,
134	capture_output=True,
135	text=True,
136	)
137
138	except _subprocess.CalledProcessError as e:	2✔
139	log.main_logger.error(	2✔
140	"Inkscape conversion failed: %s\nOutput: %s",
141	e,
142	e.output,
143	exc_info=True,
144	)
145	except OSError as e:	2✔
146	log.main_logger.error(	2✔
147	"System error running Inkscape: %s", e, exc_info=True
148	)
149
150
151	def get_file_content_as_string(	2✔
152	file_path: _pl.Path, encoding: str = "UTF-8"
153	) -> str:
154	"""Read and return the entire content of a file as a string.
155
156	Args:
157	file_path (Path): Path to the file to read
158	encoding (str, optional): File encoding to use. Defaults to "UTF-8".
159
160	Returns:
161	str: Content of the file as a string
162	"""
163	with open(file_path, "r", encoding=encoding) as file:	2✔
164	return file.read()	2✔
165
166
167	# def save_results_for_comparison(results: ResultsForComparison, path: _pl.Path) -> None:
168	# """Save ResultsForComparison data to a JSON file.
169	#
170	# This function saves monthly, hourly, step and scalar data from a ResultsForComparison
171	# object to a single JSON file.
172	#
173	# Args:
174	# results: ResultsForComparison object to save
175	# path: Path where to save the JSON file
176	#
177	# Returns:
178	# None
179	#
180	# Raises:
181	# OSError: If there's an error writing to the file
182	# """
183	# data = {
184	# 'path_to_simulations': str(results.path_to_simulations),
185	# 'monthly': {
186	# name: {
187	# 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
188	# 'columns': df.columns.tolist() if not df.empty else [],
189	# 'data': df.values.tolist() if not df.empty else []
190	# }
191	# for name, df in results.monthly.items()
192	# },
193	# 'hourly': {
194	# name: {
195	# 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
196	# 'columns': df.columns.tolist() if not df.empty else [],
197	# 'data': df.values.tolist() if not df.empty else []
198	# }
199	# for name, df in results.hourly.items()
200	# },
201	# 'step': {
202	# name: {
203	# 'index': [idx.isoformat() for idx in df.index] if not df.empty else [],
204	# 'columns': df.columns.tolist() if not df.empty else [],
205	# 'data': df.values.tolist() if not df.empty else []
206	# }
207	# for name, df in results.step.items()
208	# },
209	# 'scalar': {
210	# 'index': results.scalar.index.tolist() if not results.scalar.empty else [],
211	# 'columns': results.scalar.columns.tolist() if not results.scalar.empty else [],
212	# 'data': results.scalar.values.tolist() if not results.scalar.empty else []
213	# }
214	# }
215	#
216	# try:
217	# with open(path, 'w', encoding='utf-8') as f:
218	# _json.dump(data, f, indent=2)
219	# except OSError as e:
220	# logger.error("Error saving ResultsForComparison to JSON: %s", e, exc_info=True)
221	# raise
222	#
223	# def load_results_for_comparison(path: _pl.Path) -> ResultsForComparison:
224	# """Load ResultsForComparison data from a JSON file.
225	#
226	# This function loads monthly, hourly, step and scalar data from a JSON file
227	# and reconstructs a ResultsForComparison object.
228	#
229	# Args:
230	# path: Path to the JSON file to load
231	#
232	# Returns:
233	# ResultsForComparison: Reconstructed ResultsForComparison object
234	#
235	# Raises:
236	# OSError: If there's an error reading the file
237	# ValueError: If the file format is invalid
238	# """
239	# try:
240	# with open(path, 'r', encoding='utf-8') as f:
241	# data = _json.load(f)
242	#
243	# results = ResultsForComparison()
244	# results.path_to_simulations = _pl.Path(data['path_to_simulations'])
245	#
246	# # Load monthly data
247	# results.monthly = {
248	# name: _pd.DataFrame(
249	# data=df_data.get('data', []),
250	# index=_pd.to_datetime(df_data.get('index', [])),
251	# columns=df_data.get('columns', [])
252	# )
253	# for name, df_data in data.get('monthly', {}).items()
254	# }
255	#
256	# # Load hourly data
257	# results.hourly = {
258	# name: _pd.DataFrame(
259	# data=df_data.get('data', []),
260	# index=_pd.to_datetime(df_data.get('index', [])),
261	# columns=df_data.get('columns', [])
262	# )
263	# for name, df_data in data.get('hourly', {}).items()
264	# }
265	#
266	# # Load step data
267	# results.step = {
268	# name: _pd.DataFrame(
269	# data=df_data.get('data', []),
270	# index=_pd.to_datetime(df_data.get('index', [])),
271	# columns=df_data.get('columns', [])
272	# )
273	# for name, df_data in data.get('step', {}).items()
274	# }
275	#
276	# # Load scalar data
277	# scalar_data = data.get('scalar', {})
278	# results.scalar = _pd.DataFrame(
279	# data=scalar_data.get('data', []),
280	# index=scalar_data.get('index', []),
281	# columns=scalar_data.get('columns', [])
282	# )
283	#
284	# return results
285	#
286	# except OSError as e:
287	# logger.error("Error loading ResultsForComparison from JSON: %s", e, exc_info=True)
288	# raise
289	# except (KeyError, ValueError, _json.JSONDecodeError) as e:
290	# logger.error("Invalid ResultsForComparison JSON format: %s", e, exc_info=True)
291	# raise ValueError(f"Invalid ResultsForComparison JSON format: {str(e)}")
292
293
294	def save_to_pickle(	2✔
295	data: ds.Simulation \| ds.SimulationsData,
296	path: _pl.Path,
297	logger: _logging.Logger = log.main_logger,
298	) -> None:
299	"""Save ResultsForComparison data to a pickle file.
300
301	This function saves the entire ResultsForComparison object to a pickle file,
302	preserving all data structures and relationships.
303
304	Args:
305	data: data object to save
306	path: Path where to save the pickle file
307
308	Returns:
309	None
310
311	Raises:
312	OSError: If there's an error writing to the file
313	"""
314	try:	2✔
315	with open(path, "wb") as f:	2✔
316	_pickle.dump(data, f)	2✔
317	except OSError as e:	×
318	logger.error(	×
319	"Error saving ResultsForComparison to pickle: %s", e, exc_info=True
320	)
321	raise	×
322
323
324	def load_simulations_data_from_pickle(	2✔
325	path: _pl.Path, logger: _logging.Logger = log.main_logger
326	) -> ds.SimulationsData:
327	"""Load ResultsForComparison data from a pickle file.
328
329	This function loads a previously saved ResultsForComparison object from a pickle file.
330
331	Args:
332	path: Path to the pickle file to load
333
334	Returns:
335	ResultsForComparison: Reconstructed ResultsForComparison object
336
337	Raises:
338	OSError: If there's an error reading the file
339	pickle.UnpicklingError: If the file is corrupted or invalid
340	"""
341	try:	2✔
342	with open(path, "rb") as f:	2✔
343	simulations_data = _pickle.load(f)	2✔
344
345	# Check if it has the expected attributes of SimulationsData
346	required_attrs = {'simulations', 'scalar', 'path_to_simulations'}	2✔
347	if all(hasattr(simulations_data, attr) for attr in required_attrs):	2✔
348	return cast(ds.SimulationsData, simulations_data)	2✔
349
350	raise ValueError(	×
351	f"Loaded object is missing required SimulationsData attributes. Type: {type(simulations_data).__name__}"
352	)
353
354	except OSError as e:	×
355	logger.error(	×
356	"Error loading ResultsForComparison from pickle: %s",
357	e,
358	exc_info=True,
359	)
360	raise	×
361	except (_pickle.UnpicklingError, ValueError) as e:	×
362	logger.error(	×
363	"Invalid ResultsForComparison pickle format: %s", e, exc_info=True
364	)
365	raise	×
366
367
368	def load_simulation_from_pickle(	2✔
369	path: _pl.Path, logger: _logging.Logger = log.main_logger
370	) -> ds.Simulation:
371	try:	2✔
372	with open(path, "rb") as f:	2✔
373	simulation = _pickle.load(f)	2✔
374
375	# Check if it has the expected attributes of a Simulation
376	required_attrs = {'monthly', 'hourly', 'step', 'scalar', 'path'}	2✔
377	if all(hasattr(simulation, attr) for attr in required_attrs):	2✔
378	return cast(ds.Simulation, simulation)	2✔
379
380	raise ValueError(	×
381	f"Loaded object is missing required Simulation attributes. Type: {type(simulation).__name__}"
382	)
383	except OSError as e:	×
384	logger.error(	×
385	"Error loading Simulation from pickle: %s", e, exc_info=True
386	)
387	raise	×
388	except (_pickle.UnpicklingError, ValueError) as e:	×
389	logger.error("Invalid Simulation pickle format: %s", e, exc_info=True)	×
390	raise	×

SPF-OST / pytrnsys_process / 13394719474

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