Coverage for backend/django/core/auxiliary/services/result_object_table.py: 97%
111 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from enum import StrEnum
2from math import ceil
4from core.auxiliary.models.DataCell import DataCell
5from core.auxiliary.models.DataRow import DataRow
6from core.auxiliary.models.Scenario import Scenario
7from core.auxiliary.models.Solution import Solution
8from core.auxiliary.services.result_series_identity import (
9 ResultSeriesIndexedItemPayload,
10 result_series_identity_payload,
11)
12from pydantic import BaseModel, ConfigDict, Field
15class ResultObjectTableColumnKind(StrEnum):
16 row = "row"
17 input = "input"
18 output = "output"
19 input_output = "input_output"
22class ResultObjectTableRequest(BaseModel):
23 model_config = ConfigDict(extra="forbid")
25 scenario_id: int
26 simulation_object_id: int
27 page: int = Field(default=1, ge=1)
28 page_size: int = Field(default=20, ge=1)
31class ResultObjectTableColumn(BaseModel):
32 model_config = ConfigDict(extra="forbid")
34 key: str
35 label: str
36 unit: str | None
37 property_value_id: int | None = None
38 indexed_items: list[ResultSeriesIndexedItemPayload] = Field(
39 default_factory=list
40 )
41 kind: ResultObjectTableColumnKind
44class ResultObjectTableCell(BaseModel):
45 model_config = ConfigDict(extra="forbid")
47 column_key: str
48 value: float | None
51class ResultObjectTableRow(BaseModel):
52 model_config = ConfigDict(extra="forbid")
54 row_index: int
55 cells: list[ResultObjectTableCell]
58class ResultObjectTableResponse(BaseModel):
59 model_config = ConfigDict(extra="forbid")
61 next: str | None
62 previous: str | None
63 pages: int
64 count: int
65 page_size: int
66 row_label: str
67 columns: list[ResultObjectTableColumn]
68 rows: list[ResultObjectTableRow]
71def get_result_object_table(
72 request: ResultObjectTableRequest,
73) -> ResultObjectTableResponse:
74 """Build one object-scoped result table with merged input/output columns."""
76 scenario = Scenario.objects.get(id=request.scenario_id)
77 row_label = "Time step" if scenario.enable_dynamics else "Row"
78 rows_queryset = DataRow.objects.filter(scenario=scenario).order_by("index")
79 count = rows_queryset.count()
80 pages = ceil(count / request.page_size) if count else 1
81 page = min(request.page, pages)
82 page_start = (page - 1) * request.page_size
83 page_end = page_start + request.page_size
84 paged_rows = list(rows_queryset[page_start:page_end])
85 row_ids = [row.id for row in paged_rows]
86 row_indices = [row.index for row in paged_rows]
87 row_index_by_id = {row.id: row.index for row in paged_rows}
88 values_by_row: dict[int, dict[str, float | None]] = {
89 row_index: {} for row_index in row_indices
90 }
91 columns_by_key: dict[str, ResultObjectTableColumn] = {
92 "row_index": ResultObjectTableColumn(
93 key="row_index",
94 label=row_label,
95 unit=None,
96 kind=ResultObjectTableColumnKind.row,
97 )
98 }
99 column_order = ["row_index"]
101 _add_input_columns(
102 row_ids=row_ids,
103 row_index_by_id=row_index_by_id,
104 values_by_row=values_by_row,
105 columns_by_key=columns_by_key,
106 column_order=column_order,
107 )
108 _add_output_columns(
109 scenario=scenario,
110 simulation_object_id=request.simulation_object_id,
111 row_indices=row_indices,
112 values_by_row=values_by_row,
113 columns_by_key=columns_by_key,
114 column_order=column_order,
115 )
117 # The row index is a sticky structural column in the frontend, so response
118 # rows only carry data cells for the input/output columns.
119 data_columns = [
120 column for column in columns_by_key.values() if column.kind != "row"
121 ]
122 result_rows = [
123 ResultObjectTableRow(
124 row_index=row_index,
125 cells=[
126 ResultObjectTableCell(
127 column_key=column.key,
128 value=values_by_row[row_index].get(column.key),
129 )
130 for column in data_columns
131 ],
132 )
133 for row_index in row_indices
134 ]
136 return ResultObjectTableResponse(
137 next=_page_link(page + 1) if page < pages else None,
138 previous=_page_link(page - 1) if page > 1 else None,
139 pages=pages,
140 count=count,
141 page_size=request.page_size,
142 row_label=row_label,
143 columns=[columns_by_key[key] for key in column_order],
144 rows=result_rows,
145 )
148def _add_input_columns(
149 *,
150 row_ids: list[int],
151 row_index_by_id: dict[int, int],
152 values_by_row: dict[int, dict[str, float | None]],
153 columns_by_key: dict[str, ResultObjectTableColumn],
154 column_order: list[str],
155) -> None:
156 data_cells = (
157 DataCell.objects.filter(data_row_id__in=row_ids)
158 .select_related(
159 "data_column",
160 "data_column__property_value",
161 "data_column__property_value__property",
162 )
163 .order_by("data_column__created_at", "data_row__index")
164 )
166 for cell in data_cells:
167 data_column = cell.data_column
168 if data_column is None: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true
169 continue
170 identity = (
171 result_series_identity_payload(data_column.property_value)
172 if data_column.property_value_id
173 else None
174 )
175 column_key = _input_column_key(data_column)
176 if column_key not in columns_by_key:
177 columns_by_key[column_key] = ResultObjectTableColumn(
178 key=column_key,
179 label=identity.label if identity else data_column.name,
180 unit=identity.unit if identity else None,
181 property_value_id=identity.property_value_id if identity else None,
182 indexed_items=identity.indexed_items if identity else [],
183 kind=ResultObjectTableColumnKind.input,
184 )
185 column_order.append(column_key)
187 row_index = row_index_by_id[cell.data_row_id]
188 values_by_row[row_index][column_key] = cell.value
191def _add_output_columns(
192 *,
193 scenario: Scenario,
194 simulation_object_id: int,
195 row_indices: list[int],
196 values_by_row: dict[int, dict[str, float | None]],
197 columns_by_key: dict[str, ResultObjectTableColumn],
198 column_order: list[str],
199) -> None:
200 solutions = (
201 Solution.objects.filter(
202 scenario=scenario,
203 property__property__set__simulationObject_id=simulation_object_id,
204 )
205 .select_related("property", "property__property")
206 .prefetch_related("property__indexedItems")
207 .order_by("property__property__created_at", "property__created_at", "solve_index")
208 )
210 for solution in solutions:
211 property_value = solution.property
212 if property_value is None: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 continue
214 identity = result_series_identity_payload(property_value)
215 column_key = identity.key
216 output_label = identity.label
217 if column_key in columns_by_key:
218 if columns_by_key[column_key].kind in {
219 ResultObjectTableColumnKind.input,
220 ResultObjectTableColumnKind.input_output,
221 }:
222 columns_by_key[column_key].kind = (
223 ResultObjectTableColumnKind.input_output
224 )
225 else:
226 columns_by_key[column_key] = ResultObjectTableColumn(
227 key=column_key,
228 label=output_label,
229 unit=identity.unit,
230 property_value_id=identity.property_value_id,
231 indexed_items=identity.indexed_items,
232 kind=ResultObjectTableColumnKind.output,
233 )
234 column_order.append(column_key)
236 if scenario.enable_dynamics:
237 _add_dynamic_output_values(
238 column_key=column_key,
239 row_indices=row_indices,
240 solution=solution,
241 values_by_row=values_by_row,
242 )
243 else:
244 row_index = solution.solve_index
245 if row_index in values_by_row and solution.values:
246 values_by_row[row_index].setdefault(column_key, solution.values[0])
249def _add_dynamic_output_values(
250 *,
251 column_key: str,
252 row_indices: list[int],
253 solution: Solution,
254 values_by_row: dict[int, dict[str, float | None]],
255) -> None:
256 for row_index in row_indices:
257 if row_index >= len(solution.values):
258 continue
259 values_by_row[row_index].setdefault(column_key, solution.values[row_index])
262def _input_column_key(data_column) -> str:
263 if data_column.property_value_id:
264 return result_series_identity_payload(data_column.property_value).key
265 return f"label:{_normalize_column_label(data_column.name)}"
268def _normalize_column_label(label: str) -> str:
269 return " ".join(label.casefold().split())
272def _page_link(page: int) -> str:
273 return f"page={page}"