Coverage for backend/django/core/auxiliary/services/result_summary/summaries.py: 94%
46 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from common.models.notifications.payloads import ResultSummaryCalculationError
2from core.auxiliary.models.Scenario import Scenario
3from core.auxiliary.services.result_series_identity import (
4 result_series_identity_payload,
5)
6from core.auxiliary.services.result_summary.cache_payloads import (
7 CachedResultSummary,
8 CachedResultSummaryBuild,
9 ResultSummaryCachePayload,
10)
11from core.auxiliary.services.result_summary.contracts import (
12 ResultSummary,
13 ResultSummaryRequest,
14)
15from core.auxiliary.services.result_summary.histograms import (
16 cached_histogram_variants,
17 histogram_bins,
18)
19from core.auxiliary.services.result_summary.series_queries import (
20 result_summary_values,
21 safe_property_value_key,
22 steady_state_property_values,
23 steady_state_result_series,
24)
25from core.auxiliary.services.result_summary.stats import (
26 finite_values,
27 metric_targets,
28 result_stats,
29)
32def get_result_summaries(request: ResultSummaryRequest) -> list[ResultSummary]:
33 """Build per-property result statistics and histogram bins for one unit op."""
35 scenario = Scenario.objects.get(id=request.scenario_id)
36 if not scenario.enable_dynamics:
37 return _steady_state_result_summaries(
38 request,
39 flowsheet_id=scenario.flowsheet_state.flowsheet_id,
40 )
42 summaries: list[ResultSummary] = []
43 for result_values in result_summary_values(
44 scenario_id=request.scenario_id,
45 simulation_object_id=request.simulation_object_id,
46 ):
47 finite_result_values = finite_values(result_values.values)
48 stats = result_stats(finite_result_values, len(result_values.values))
49 summaries.append(
50 ResultSummary(
51 key=result_values.key,
52 label=result_values.label,
53 unit=result_values.unit,
54 property_value_id=result_values.property_value_id,
55 indexed_items=result_values.indexed_items,
56 stats=stats,
57 metric_targets=metric_targets(finite_result_values, stats),
58 histogram_bins=histogram_bins(
59 finite_result_values,
60 request.histogram_options_for(result_values.key),
61 ),
62 )
63 )
64 return summaries
67def build_cached_result_summaries(
68 *,
69 scenario_id: int,
70 simulation_object_id: int,
71 source_task_id: int,
72 flowsheet_id: int,
73) -> ResultSummaryCachePayload:
74 """Build all supported histogram variants for one MSS scenario/object."""
76 return build_cached_result_summaries_with_errors(
77 scenario_id=scenario_id,
78 simulation_object_id=simulation_object_id,
79 source_task_id=source_task_id,
80 flowsheet_id=flowsheet_id,
81 ).payload
84def build_cached_result_summaries_with_errors(
85 *,
86 scenario_id: int,
87 simulation_object_id: int,
88 source_task_id: int,
89 flowsheet_id: int,
90) -> CachedResultSummaryBuild:
91 """Build cached summaries while preserving per-result item failures."""
93 property_values = steady_state_property_values(
94 scenario_id=scenario_id,
95 simulation_object_id=simulation_object_id,
96 flowsheet_id=flowsheet_id,
97 )
98 property_value_ids = [property_value.id for property_value in property_values]
99 series_by_property_id = steady_state_result_series(
100 scenario_id=scenario_id,
101 flowsheet_id=flowsheet_id,
102 property_value_ids=property_value_ids,
103 )
104 summaries: list[CachedResultSummary] = []
105 errors: list[ResultSummaryCalculationError] = []
106 for property_value in property_values:
107 try:
108 identity = result_series_identity_payload(property_value)
109 series = series_by_property_id[property_value.id]
110 stats = result_stats(series.values, series.total_count)
111 summaries.append(
112 CachedResultSummary(
113 key=identity.key,
114 label=identity.label,
115 unit=identity.unit,
116 property_value_id=identity.property_value_id,
117 indexed_items=identity.indexed_items,
118 stats=stats,
119 metric_targets=metric_targets(series.values, stats),
120 histogram_bins=cached_histogram_variants(series.values),
121 )
122 )
123 except Exception as error:
124 errors.append(
125 ResultSummaryCalculationError(
126 message=(
127 "One result could not be included in the aggregate summary."
128 ),
129 cause=error.__class__.__name__,
130 simulation_object_id=simulation_object_id,
131 result_key=safe_property_value_key(property_value),
132 property_value_id=property_value.id,
133 )
134 )
136 return CachedResultSummaryBuild(
137 payload=ResultSummaryCachePayload(
138 scenario_id=scenario_id,
139 simulation_object_id=simulation_object_id,
140 source_task_id=source_task_id,
141 summaries=summaries,
142 ),
143 errors=errors,
144 )
147def _steady_state_result_summaries(
148 request: ResultSummaryRequest,
149 *,
150 flowsheet_id: int,
151) -> list[ResultSummary]:
152 property_values = steady_state_property_values(
153 scenario_id=request.scenario_id,
154 simulation_object_id=request.simulation_object_id,
155 flowsheet_id=flowsheet_id,
156 )
157 property_value_ids = [property_value.id for property_value in property_values]
158 series_by_property_id = steady_state_result_series(
159 scenario_id=request.scenario_id,
160 flowsheet_id=flowsheet_id,
161 property_value_ids=property_value_ids,
162 )
163 summaries: list[ResultSummary] = []
164 for property_value in property_values:
165 identity = result_series_identity_payload(property_value)
166 series = series_by_property_id[property_value.id]
167 stats = result_stats(series.values, series.total_count)
168 summaries.append(
169 ResultSummary(
170 key=identity.key,
171 label=identity.label,
172 unit=identity.unit,
173 property_value_id=identity.property_value_id,
174 indexed_items=identity.indexed_items,
175 stats=stats,
176 metric_targets=metric_targets(series.values, stats),
177 histogram_bins=histogram_bins(
178 series.values,
179 request.histogram_options_for(identity.key),
180 ),
181 )
182 )
183 return summaries