Coverage for backend/django/core/auxiliary/services/result_summary/series_queries.py: 87%
71 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from collections.abc import Iterator
2from math import isfinite
4from core.auxiliary.models.PropertyValue import PropertyValue
5from core.auxiliary.models.Scenario import Scenario
6from core.auxiliary.models.Solution import Solution
7from core.auxiliary.services.result_series_identity import (
8 ResultSeriesIndexedItemPayload,
9 result_series_identity_payload,
10)
11from core.auxiliary.services.result_summary.stats import FiniteResultValuePoint
12from django.db import connection
13from pydantic import BaseModel, ConfigDict
16class SteadyStateResultSeries:
17 def __init__(self) -> None:
18 self.total_count = 0
19 self.values: list[FiniteResultValuePoint] = []
22class ResultValuePoint(BaseModel):
23 model_config = ConfigDict(extra="forbid")
25 row_index: int
26 value: float | None
29class ResultSummaryValues(BaseModel):
30 model_config = ConfigDict(extra="forbid")
32 key: str
33 label: str
34 unit: str | None
35 property_value_id: int
36 indexed_items: list[ResultSeriesIndexedItemPayload]
37 values: list[ResultValuePoint]
40def steady_state_property_values(
41 *, scenario_id: int, simulation_object_id: int, flowsheet_id: int
42) -> list[PropertyValue]:
43 property_value_ids = (
44 Solution.objects.filter(
45 flowsheet_state__flowsheet_id=flowsheet_id,
46 scenario_id=scenario_id,
47 property__property__set__simulationObject_id=simulation_object_id,
48 )
49 .order_by()
50 .values_list("property_id", flat=True)
51 .distinct()
52 )
53 return list(
54 PropertyValue.objects.filter(id__in=property_value_ids)
55 .select_related("property")
56 .prefetch_related("indexedItems")
57 .order_by("property__created_at", "created_at")
58 )
61def steady_state_result_series(
62 *, scenario_id: int, flowsheet_id: int, property_value_ids: list[int]
63) -> dict[int, SteadyStateResultSeries]:
64 series_by_property_id = {
65 property_value_id: SteadyStateResultSeries()
66 for property_value_id in property_value_ids
67 }
68 if not property_value_ids: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 return series_by_property_id
71 for property_value_id, solve_index, value in steady_state_result_rows(
72 scenario_id=scenario_id,
73 flowsheet_id=flowsheet_id,
74 property_value_ids=property_value_ids,
75 ):
76 series = series_by_property_id[property_value_id]
77 # Keep total_count as the number of returned result rows, not just the
78 # finite values, so the UI can report omitted null/non-numeric values.
79 series.total_count += 1
80 if value is None:
81 continue
82 try:
83 numeric_value = float(value)
84 except (TypeError, ValueError):
85 continue
86 if isfinite(numeric_value):
87 series.values.append(
88 FiniteResultValuePoint(
89 row_index=solve_index,
90 value=numeric_value,
91 )
92 )
93 return series_by_property_id
96def steady_state_result_rows(
97 *, scenario_id: int, flowsheet_id: int, property_value_ids: list[int]
98) -> Iterator[tuple[int, int, float | None]]:
99 flowsheet_state_id = (
100 Scenario.objects.only("flowsheet_state_id")
101 .get(
102 pk=scenario_id,
103 flowsheet_state__flowsheet_id=flowsheet_id,
104 )
105 .flowsheet_state_id
106 )
107 with connection.cursor() as cursor:
108 # This hot path intentionally uses a parameterized raw cursor instead of
109 # the ORM: Django row conversion dominated summary generation for large
110 # MSS result sets. Keep values in the params list rather than formatting
111 # them into the SQL string so the database driver handles escaping.
112 # Raw SQL bypasses AccessControlManager, so keep the authorized
113 # scenario flowsheet in the predicate as an explicit data boundary.
114 cursor.execute(
115 """
116 SELECT property_id, solve_index, "values"[1] AS result_value
117 FROM core_auxiliary_solution
118 WHERE flowsheet_state_id = %s AND scenario_id = %s AND property_id = ANY(%s)
119 ORDER BY property_id, solve_index
120 """,
121 [flowsheet_state_id, scenario_id, property_value_ids],
122 )
123 while True:
124 rows = cursor.fetchmany(20000)
125 if not rows:
126 break
127 yield from rows
130def safe_property_value_key(property_value: PropertyValue) -> str | None:
131 try:
132 return result_series_identity_payload(property_value).key
133 except Exception:
134 return None
137def result_summary_values(
138 *, scenario_id: int | str, simulation_object_id: int | str
139) -> list[ResultSummaryValues]:
140 scenario = Scenario.objects.get(id=scenario_id)
141 solutions = (
142 Solution.objects.filter(
143 scenario_id=scenario_id,
144 property__property__set__simulationObject_id=simulation_object_id,
145 )
146 .select_related("property", "property__property")
147 .prefetch_related("property__indexedItems")
148 .order_by("solve_index")
149 )
151 grouped_values: dict[str, ResultSummaryValues] = {}
152 for solution in solutions:
153 identity = result_series_identity_payload(solution.property)
154 key = identity.key
155 if key not in grouped_values: 155 ↛ 164line 155 didn't jump to line 164 because the condition on line 155 was always true
156 grouped_values[key] = ResultSummaryValues(
157 key=key,
158 label=identity.label,
159 unit=identity.unit,
160 property_value_id=identity.property_value_id,
161 indexed_items=identity.indexed_items,
162 values=[],
163 )
164 if scenario.enable_dynamics: 164 ↛ 170line 164 didn't jump to line 170 because the condition on line 164 was always true
165 grouped_values[key].values.extend(
166 ResultValuePoint(row_index=index, value=value)
167 for index, value in enumerate(solution.values)
168 )
169 else:
170 grouped_values[key].values.append(
171 ResultValuePoint(
172 row_index=solution.solve_index,
173 value=solution.values[0] if solution.values else None,
174 )
175 )
177 return list(grouped_values.values())