16748736500

Committed 05 Aug 2025 11:27AM UTC coverage: 95.338% (-0.6%) from 95.968%

Build Type

github

Committed by ahobeost

Commit Message

Use caplog instead of reading file.

Pull Request Pull Request #126: 125 bug step file not read when step used with type 25

Run Details

6 of 6 new or added lines in 2 files covered. (100.0%)

8 existing lines in 1 file now uncovered.

1186 of 1244 relevant lines covered (95.34%)

0.95 hits per line

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

80.95

/examples/ready_to_run/processing_example.py

import pathlib as _pl

import matplotlib.pyplot as _plt

from pytrnsys_process import api


# Your processing steps are writen inside functions.
# Which need to be defined before your scripts entry point.
# Define your processing steps at the top of the script.
def processing_of_monthly_data(simulation: api.Simulation):
    # create a bar chart using monthly data
    monthly_df = simulation.monthly
    columns_to_plot = ["QSnk60P", "QSnk60PauxCondSwitch_kW"]
    fig, ax = api.bar_chart(
        monthly_df,
        columns_to_plot,
    )

    # modify label for the y axis
    ax.set_ylabel("Power [kW]")

    # This is also done by api.save_plot, but if you want to see your plot before saving it.
    # For example with _plt.show(). Calling this function is required, to make the plot look as expected.
    # _plt.tight_layout()

    # create plots folder inside simulation directory and save plot in configured formats
    api.export_plots_in_configured_formats(
        fig, simulation.path, "monthly-bar-chart"
    )


def processing_of_hourly_data(simulation: api.Simulation):
    # create line plot using hourly data
    fig, ax = api.line_plot(simulation.hourly, ["QSrc1TIn", "QSrc1TOut"])

    # Here you can modify anything to do with your plot, according to the matplotlib.pyplot standard
    ax.set_ylabel("Temperature [°C]")
    # make label fit inside your plot
    # _plt.tight_layout()

    # create plots folder inside simulation directory and save plot in configured formats
    api.export_plots_in_configured_formats(
        fig, simulation.path, "hourly-line-plot"
    )


def processing_for_histogram(simulation: api.Simulation):
    # create histogram using hourly data
    api.histogram(simulation.hourly, ["QSrc1TIn"])

    # if you don't want to export your plot, you can also just show it by calling this function
    _plt.show()


def calc_total_yearly_demand_per_sim(
    simulations_data: api.SimulationsData,
) -> api.SimulationsData:
    # Columns to sum
    demand_columns = [
        "qSysOut_QSnk131Demand",
        "qSysOut_QSnk183Demand",
        "qSysOut_QSnk191Demand",
        "qSysOut_QSnk225Demand",
        "qSysOut_QSnk243Demand",
        "qSysOut_QSnk266Demand",
        "qSysOut_QSnk322Demand",
        "qSysOut_QSnk335Demand",
        "qSysOut_QSnk358Demand",
        "qSysOut_QSnk417Demand",
        "qSysOut_QSnk448Demand",
        "qSysOut_QSnk469Demand",
        "qSysOut_QSnk488Demand",
        "qSysOut_QSnk524Demand",
        "qSysOut_QSnk539Demand",
        "qSysOut_QSnk558Demand",
        "qSysOut_QSnk579Demand",
        "qSysOut_QSnk60Demand",
        "qSysOut_QSnk85Demand",
    ]
    # Conversion to gwh
    kwh_to_gwh = 1e-6
    # Iterate through each simulation
    for sim_name, sim in simulations_data.simulations.items():
        # Access the monthly data
        monthly_df = sim.monthly
        # Calculate total monthly demand in GWh (sum columns and convert from kWh)
        monthly_df["total_demand_GWh"] = (
            monthly_df[demand_columns].sum(axis=1) * kwh_to_gwh
        )

        # Sum Jan-December values (skip first 2 months - typically Nov/Dec)
        # Wrap in int to get rid of decimal places
        yearly_total = int(monthly_df["total_demand_GWh"].iloc[2::].sum())

        # Store result in scalar data for comparison
        simulations_data.scalar.loc[sim_name, "yearly_demand_GWh"] = (
            yearly_total
        )
    return simulations_data


def calc_v_ice_ratio_max_per_sim(
    simulations_data: api.SimulationsData,
) -> api.SimulationsData:
    # Iterate through each simulation's hourly data
    for sim_name, sim in simulations_data.simulations.items():
        # Access hourly data
        hourly_data = sim.hourly
        # Get maximum VIceRatio value from hourly timeseries
        max_ice_ratio = hourly_data["VIceRatio"].max()
        # Store result in scalar data table using simulation name as index
        simulations_data.scalar.loc[sim_name, "VIceRatioMax"] = max_ice_ratio
    return simulations_data


def comparison_of_scalar_data(
    simulations_data: api.SimulationsData,
):
    # Calculate and add two new metrics to the results:
    # 1. Total yearly energy demand in GWh
    # 2. Maximum ice storage ratio from hourly data

    # Grouping steps will ensure that the failure of a dependent step will cause the whole process to fail.
    simulations_data = calc_total_yearly_demand_per_sim(simulations_data)
    simulations_data = calc_v_ice_ratio_max_per_sim(simulations_data)

    # Create a scatter plot for comparison comparing:
    fig, _ = api.scalar_compare_plot(
        simulations_data.scalar,
        "VIceSscaled",
        "VIceRatioMax",
        group_by_color="yearly_demand_GWh",
        group_by_marker="ratioDHWtoSH_allSinks",
    )

    # save plots in the results folder in the configured formats
    api.export_plots_in_configured_formats(
        fig,
        simulations_data.path_to_simulations,
        "compare-plot",
        plots_folder_name="",
    )


def main():
    # This is the entry point to your script.
    # Here, you can decide how you would like to run your processing steps.
    # In the example below we run both processing steps on a whole result set.
    # Or a single processing step on a single simulation.
    # Make sure to remove the lines you do not want to run.

    # bundle the steps into a list
    processing_scenarios = [
        processing_of_monthly_data,
        processing_of_hourly_data,
    ]

    # run the scenarios on a whole result using multiple processes
    simulations_data = api.process_whole_result_set_parallel(
        _pl.Path(api.REPO_ROOT / "examples/ready_to_run/data/results"),
        processing_scenarios,
    )

    # using the data from processing multiple simulations, run your comparisons steps
    api.do_comparison(comparison_of_scalar_data, simulations_data)

    # run the single scenario on a single simulation
    (
        api.process_single_simulation(
            _pl.Path(
                api.REPO_ROOT
                / "examples/ready_to_run/data/results/complete-0-SnkScale0.8000-StoreScale10"
            ),
            # ===============================================================
            processing_for_histogram,
            # do not add round brackets when linking your processing step
            # processing_for_histogram AND NOT processing_for_histogram()
            # ===============================================================
        )
    )


# This is required since we would like to run the script directly
if __name__ == "__main__":
    main()  # pragma: no cover

1	import pathlib as _pl	1✔
2
3	import matplotlib.pyplot as _plt	1✔
4
5	from pytrnsys_process import api	1✔
6
7
8	# Your processing steps are writen inside functions.
9	# Which need to be defined before your scripts entry point.
10	# Define your processing steps at the top of the script.
11	def processing_of_monthly_data(simulation: api.Simulation):	1✔
12	# create a bar chart using monthly data
UNCOV 13	monthly_df = simulation.monthly	×
UNCOV 14	columns_to_plot = ["QSnk60P", "QSnk60PauxCondSwitch_kW"]	×
UNCOV 15	fig, ax = api.bar_chart(	×
16	monthly_df,
17	columns_to_plot,
18	)
19
20	# modify label for the y axis
UNCOV 21	ax.set_ylabel("Power [kW]")	×
22
23	# This is also done by api.save_plot, but if you want to see your plot before saving it.
24	# For example with _plt.show(). Calling this function is required, to make the plot look as expected.
25	# _plt.tight_layout()
26
27	# create plots folder inside simulation directory and save plot in configured formats
UNCOV 28	api.export_plots_in_configured_formats(	×
29	fig, simulation.path, "monthly-bar-chart"
30	)
31
32
33	def processing_of_hourly_data(simulation: api.Simulation):	1✔
34	# create line plot using hourly data
UNCOV 35	fig, ax = api.line_plot(simulation.hourly, ["QSrc1TIn", "QSrc1TOut"])	×
36
37	# Here you can modify anything to do with your plot, according to the matplotlib.pyplot standard
UNCOV 38	ax.set_ylabel("Temperature [°C]")	×
39	# make label fit inside your plot
40	# _plt.tight_layout()
41
42	# create plots folder inside simulation directory and save plot in configured formats
UNCOV 43	api.export_plots_in_configured_formats(	×
44	fig, simulation.path, "hourly-line-plot"
45	)
46
47
48	def processing_for_histogram(simulation: api.Simulation):	1✔
49	# create histogram using hourly data
50	api.histogram(simulation.hourly, ["QSrc1TIn"])	1✔
51
52	# if you don't want to export your plot, you can also just show it by calling this function
53	_plt.show()	1✔
54
55
56	def calc_total_yearly_demand_per_sim(	1✔
57	simulations_data: api.SimulationsData,
58	) -> api.SimulationsData:
59	# Columns to sum
60	demand_columns = [	1✔
61	"qSysOut_QSnk131Demand",
62	"qSysOut_QSnk183Demand",
63	"qSysOut_QSnk191Demand",
64	"qSysOut_QSnk225Demand",
65	"qSysOut_QSnk243Demand",
66	"qSysOut_QSnk266Demand",
67	"qSysOut_QSnk322Demand",
68	"qSysOut_QSnk335Demand",
69	"qSysOut_QSnk358Demand",
70	"qSysOut_QSnk417Demand",
71	"qSysOut_QSnk448Demand",
72	"qSysOut_QSnk469Demand",
73	"qSysOut_QSnk488Demand",
74	"qSysOut_QSnk524Demand",
75	"qSysOut_QSnk539Demand",
76	"qSysOut_QSnk558Demand",
77	"qSysOut_QSnk579Demand",
78	"qSysOut_QSnk60Demand",
79	"qSysOut_QSnk85Demand",
80	]
81	# Conversion to gwh
82	kwh_to_gwh = 1e-6	1✔
83	# Iterate through each simulation
84	for sim_name, sim in simulations_data.simulations.items():	1✔
85	# Access the monthly data
86	monthly_df = sim.monthly	1✔
87	# Calculate total monthly demand in GWh (sum columns and convert from kWh)
88	monthly_df["total_demand_GWh"] = (	1✔
89	monthly_df[demand_columns].sum(axis=1) * kwh_to_gwh
90	)
91
92	# Sum Jan-December values (skip first 2 months - typically Nov/Dec)
93	# Wrap in int to get rid of decimal places
94	yearly_total = int(monthly_df["total_demand_GWh"].iloc[2::].sum())	1✔
95
96	# Store result in scalar data for comparison
97	simulations_data.scalar.loc[sim_name, "yearly_demand_GWh"] = (	1✔
98	yearly_total
99	)
100	return simulations_data	1✔
101
102
103	def calc_v_ice_ratio_max_per_sim(	1✔
104	simulations_data: api.SimulationsData,
105	) -> api.SimulationsData:
106	# Iterate through each simulation's hourly data
107	for sim_name, sim in simulations_data.simulations.items():	1✔
108	# Access hourly data
109	hourly_data = sim.hourly	1✔
110	# Get maximum VIceRatio value from hourly timeseries
111	max_ice_ratio = hourly_data["VIceRatio"].max()	1✔
112	# Store result in scalar data table using simulation name as index
113	simulations_data.scalar.loc[sim_name, "VIceRatioMax"] = max_ice_ratio	1✔
114	return simulations_data	1✔
115
116
117	def comparison_of_scalar_data(	1✔
118	simulations_data: api.SimulationsData,
119	):
120	# Calculate and add two new metrics to the results:
121	# 1. Total yearly energy demand in GWh
122	# 2. Maximum ice storage ratio from hourly data
123
124	# Grouping steps will ensure that the failure of a dependent step will cause the whole process to fail.
125	simulations_data = calc_total_yearly_demand_per_sim(simulations_data)	1✔
126	simulations_data = calc_v_ice_ratio_max_per_sim(simulations_data)	1✔
127
128	# Create a scatter plot for comparison comparing:
129	fig, _ = api.scalar_compare_plot(	1✔
130	simulations_data.scalar,
131	"VIceSscaled",
132	"VIceRatioMax",
133	group_by_color="yearly_demand_GWh",
134	group_by_marker="ratioDHWtoSH_allSinks",
135	)
136
137	# save plots in the results folder in the configured formats
138	api.export_plots_in_configured_formats(	1✔
139	fig,
140	simulations_data.path_to_simulations,
141	"compare-plot",
142	plots_folder_name="",
143	)
144
145
146	def main():	1✔
147	# This is the entry point to your script.
148	# Here, you can decide how you would like to run your processing steps.
149	# In the example below we run both processing steps on a whole result set.
150	# Or a single processing step on a single simulation.
151	# Make sure to remove the lines you do not want to run.
152
153	# bundle the steps into a list
154	processing_scenarios = [	1✔
155	processing_of_monthly_data,
156	processing_of_hourly_data,
157	]
158
159	# run the scenarios on a whole result using multiple processes
160	simulations_data = api.process_whole_result_set_parallel(	1✔
161	_pl.Path(api.REPO_ROOT / "examples/ready_to_run/data/results"),
162	processing_scenarios,
163	)
164
165	# using the data from processing multiple simulations, run your comparisons steps
166	api.do_comparison(comparison_of_scalar_data, simulations_data)	1✔
167
168	# run the single scenario on a single simulation
169	(	1✔
170	api.process_single_simulation(
171	_pl.Path(
172	api.REPO_ROOT
173	/ "examples/ready_to_run/data/results/complete-0-SnkScale0.8000-StoreScale10"
174	),
175	# ===============================================================
176	processing_for_histogram,
177	# do not add round brackets when linking your processing step
178	# processing_for_histogram AND NOT processing_for_histogram()
179	# ===============================================================
180	)
181	)
182
183
184	# This is required since we would like to run the script directly
185	if __name__ == "__main__":	1✔
186	main() # pragma: no cover

SPF-OST / pytrnsys_process / 16748736500

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