Coverage for backend/django/core/auxiliary/services/mss_input_data.py: 94%
83 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from collections.abc import Iterable
2from dataclasses import dataclass
3from enum import StrEnum
5from django.utils import timezone
7from core.auxiliary.models.DataCell import DataCell
8from core.auxiliary.models.DataRow import DataRow
9from core.auxiliary.models.Scenario import Scenario
10from core.auxiliary.services.scenario_time_series import is_mss_time_series
11from core.auxiliary.services.result_summary.cache import (
12 ResultSummaryCacheInvalidationReason,
13 invalidate_result_summary_cache_for_scenario,
14)
16CELL_BULK_BATCH_SIZE = 5000
17ROW_BULK_BATCH_SIZE = 1000
20class MssInputSource(StrEnum):
21 csv_import = "csv_import"
22 legacy_upload = "legacy_upload"
23 legacy_delete = "legacy_delete"
24 mode_switch = "mode_switch"
25 parameter_sweep = "parameter_sweep"
28@dataclass(frozen=True)
29class MssInputCell:
30 data_column_id: int
31 value: float
34@dataclass(frozen=True)
35class MssInputRow:
36 index: int
37 cells: list[MssInputCell]
40def replace_mss_input_rows(
41 *,
42 scenario: Scenario,
43 rows: Iterable[MssInputRow],
44 source: MssInputSource,
45) -> None:
46 """Replace all MSS input rows/cells for a scenario and clear stale summaries."""
48 invalidate_result_summary_cache_for_scenario(
49 scenario_id=scenario.id,
50 reason=ResultSummaryCacheInvalidationReason.mss_input_replaced,
51 )
52 _mark_economics_schedule_inputs_changed(scenario_ids=[scenario.id])
53 DataRow.objects.filter(scenario=scenario).delete()
55 row_batch: list[MssInputRow] = []
56 for row in rows:
57 row_batch.append(row)
58 if len(row_batch) >= ROW_BULK_BATCH_SIZE:
59 _create_mss_input_row_batch(scenario=scenario, rows=row_batch)
60 row_batch.clear()
61 if row_batch:
62 _create_mss_input_row_batch(scenario=scenario, rows=row_batch)
65def _create_mss_input_row_batch(
66 *,
67 scenario: Scenario,
68 rows: list[MssInputRow],
69) -> None:
70 """Persist one bounded MSS input-row batch and its cells."""
72 data_rows = [
73 DataRow(
74 index=row.index,
75 flowsheet_state=scenario.flowsheet_state,
76 scenario=scenario,
77 )
78 for row in rows
79 ]
80 DataRow.objects.bulk_create(data_rows, batch_size=ROW_BULK_BATCH_SIZE)
81 row_indexes = [row.index for row in rows]
82 data_rows_by_index = {
83 row.index: row
84 for row in DataRow.objects.filter(
85 scenario=scenario,
86 index__in=row_indexes,
87 ).order_by("index")
88 }
90 cells: list[DataCell] = []
91 for row in rows:
92 data_row = data_rows_by_index[row.index]
93 for cell in row.cells:
94 cells.append(
95 DataCell(
96 flowsheet_state=scenario.flowsheet_state,
97 data_column_id=cell.data_column_id,
98 data_row=data_row,
99 value=cell.value,
100 )
101 )
102 if len(cells) >= CELL_BULK_BATCH_SIZE:
103 DataCell.objects.bulk_create(cells, batch_size=CELL_BULK_BATCH_SIZE)
104 cells.clear()
105 if cells: 105 ↛ exitline 105 didn't return from function '_create_mss_input_row_batch' because the condition on line 105 was always true
106 DataCell.objects.bulk_create(cells, batch_size=CELL_BULK_BATCH_SIZE)
109def clear_mss_input_rows(*, scenario: Scenario, source: MssInputSource) -> None:
110 """Delete all MSS input rows/cells for a scenario and clear stale summaries."""
112 invalidate_result_summary_cache_for_scenario(
113 scenario_id=scenario.id,
114 reason=ResultSummaryCacheInvalidationReason.mss_input_cleared,
115 )
116 _mark_economics_schedule_inputs_changed(scenario_ids=[scenario.id])
117 DataRow.objects.filter(scenario=scenario).delete()
120def mark_mss_input_rows_changed(*, data_row_ids: Iterable[int]) -> None:
121 """Mark direct MSS input-cell edits as needing a fresh solve."""
123 row_ids = {row_id for row_id in data_row_ids if row_id is not None}
124 if not row_ids: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 return
127 rows = list(
128 DataRow.objects.filter(id__in=row_ids)
129 .select_related("scenario")
130 .only(
131 "id",
132 "scenario_id",
133 "scenario__state_name",
134 "scenario__mss_time_series_enabled",
135 "scenario__enable_dynamics",
136 )
137 )
138 mss_rows = [
139 row
140 for row in rows
141 if row.scenario_id is not None and is_mss_time_series(row.scenario)
142 ]
143 if not mss_rows: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 return
146 DataRow.objects.filter(id__in=[row.id for row in mss_rows]).update(created_at=timezone.now())
147 for scenario_id in {row.scenario_id for row in mss_rows}:
148 invalidate_result_summary_cache_for_scenario(
149 scenario_id=scenario_id,
150 reason=ResultSummaryCacheInvalidationReason.mss_input_cell_changed,
151 )
152 _mark_economics_schedule_inputs_changed(scenario_ids=[row.scenario_id for row in mss_rows])
155def mark_mss_input_scenarios_changed(*, scenario_ids: Iterable[int]) -> None:
156 """Mark direct MSS input-column edits as needing a fresh solve."""
158 scenario_id_set = {scenario_id for scenario_id in scenario_ids if scenario_id is not None}
159 if not scenario_id_set: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 return
162 scenarios = list(
163 Scenario.objects.filter(id__in=scenario_id_set).only(
164 "id",
165 "state_name",
166 "mss_time_series_enabled",
167 "enable_dynamics",
168 )
169 )
170 mss_scenario_ids = [scenario.id for scenario in scenarios if is_mss_time_series(scenario)]
171 if not mss_scenario_ids:
172 return
174 DataRow.objects.filter(scenario_id__in=mss_scenario_ids).update(created_at=timezone.now())
175 for scenario_id in mss_scenario_ids:
176 invalidate_result_summary_cache_for_scenario(
177 scenario_id=scenario_id,
178 reason=ResultSummaryCacheInvalidationReason.mss_input_column_changed,
179 )
180 _mark_economics_schedule_inputs_changed(scenario_ids=mss_scenario_ids)
183def _mark_economics_schedule_inputs_changed(*, scenario_ids: Iterable[int]) -> None:
184 from Economics.scheduling.events import schedule_scenario_inputs_changed
186 schedule_scenario_inputs_changed(scenario_ids=scenario_ids)