Coverage for backend/django/Economics/results/services/comparison/charts.py: 97%
147 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Build comparison-chart payloads from persisted result chart datasets."""
3from __future__ import annotations
5from copy import deepcopy
6from typing import Any
8from Economics.results.models import EconomicsChartDataset
9from Economics.results.services.chart_datasets import (
10 CHART_CAPEX_BREAKDOWN,
11 CHART_CASH_FLOW_NPV,
12 CHART_OPEX_BREAKDOWN,
13 CHART_OPERATING_COST_CUMULATIVE,
14 CHART_OPERATING_COST_PROFILE,
15)
17from .contracts import (
18 ComparisonChartDatasetPayload,
19 ComparisonTarget,
20 JsonValue,
21 ResultContext,
22)
25COMPARISON_CHART_CASH_FLOW = "comparison_cash_flow_npv"
26COMPARISON_CHART_COST_BREAKDOWN = "comparison_cost_breakdown"
27COMPARISON_CHART_OPERATING_CUMULATIVE = "comparison_operating_cost_cumulative"
28COMPARISON_CHART_OPERATING_PROFILE = "comparison_operating_cost_profile"
30_CASH_FLOW_METRIC_LABELS = {
31 "cumulative_discounted_cash_flow": "Cumulative discounted cash flow",
32 "annual_net_cash_flow": "Annual net cash flow",
33}
34_COST_BREAKDOWN_METRIC_LABELS = {
35 CHART_CAPEX_BREAKDOWN: "Capital cost",
36 CHART_OPEX_BREAKDOWN: "Operating cost",
37}
38_SOURCE_CHART_KEYS = (
39 CHART_CASH_FLOW_NPV,
40 CHART_CAPEX_BREAKDOWN,
41 CHART_OPEX_BREAKDOWN,
42 CHART_OPERATING_COST_CUMULATIVE,
43 CHART_OPERATING_COST_PROFILE,
44)
47def comparison_chart_datasets(
48 *,
49 studies: list[ComparisonTarget],
50 result_contexts: dict[int | None, ResultContext | None],
51) -> list[ComparisonChartDatasetPayload]:
52 """Return comparison-overlay chart datasets using each study's current chart rows."""
53 datasets_by_run = _datasets_by_run(result_contexts)
54 chart_datasets = [
55 _cash_flow_overlay_dataset(
56 studies=studies,
57 result_contexts=result_contexts,
58 datasets_by_run=datasets_by_run,
59 ),
60 _cost_breakdown_dataset(
61 studies=studies,
62 result_contexts=result_contexts,
63 datasets_by_run=datasets_by_run,
64 ),
65 _operating_timeline_overlay_dataset(
66 studies=studies,
67 result_contexts=result_contexts,
68 datasets_by_run=datasets_by_run,
69 source_chart_key=CHART_OPERATING_COST_CUMULATIVE,
70 comparison_chart_key=COMPARISON_CHART_OPERATING_CUMULATIVE,
71 title="Cumulative Operating Cost",
72 dataset_id=-3,
73 ),
74 _operating_timeline_overlay_dataset(
75 studies=studies,
76 result_contexts=result_contexts,
77 datasets_by_run=datasets_by_run,
78 source_chart_key=CHART_OPERATING_COST_PROFILE,
79 comparison_chart_key=COMPARISON_CHART_OPERATING_PROFILE,
80 title="Operating Cost Profile",
81 dataset_id=-4,
82 ),
83 ]
84 return [dataset for dataset in chart_datasets if dataset is not None]
87def _datasets_by_run(
88 result_contexts: dict[int | None, ResultContext | None],
89) -> dict[int, dict[str, EconomicsChartDataset]]:
90 run_ids = [
91 context.run.pk
92 for context in result_contexts.values()
93 if context is not None and context.run.pk is not None
94 ]
95 datasets: dict[int, dict[str, EconomicsChartDataset]] = {}
96 if not run_ids:
97 return datasets
98 # Result contexts are loaded per study only after access to that study's
99 # flowsheet is checked. Use the base manager here so chart rows from
100 # selected studies in other authorized flowsheets are not filtered out by
101 # the active request flowsheet context.
102 for dataset in EconomicsChartDataset._base_manager.filter(
103 result_run_id__in=run_ids,
104 chart_key__in=_SOURCE_CHART_KEYS,
105 ).order_by("result_run_id", "chart_key"):
106 datasets.setdefault(dataset.result_run_id, {})[dataset.chart_key] = dataset
107 return datasets
110def _cash_flow_overlay_dataset(
111 *,
112 studies: list[ComparisonTarget],
113 result_contexts: dict[int | None, ResultContext | None],
114 datasets_by_run: dict[int, dict[str, EconomicsChartDataset]],
115) -> ComparisonChartDatasetPayload | None:
116 series = []
117 source_row_keys: list[str] = []
118 warning_refs: list[dict[str, JsonValue]] = []
119 target_labels = _target_labels(studies)
120 for target in studies:
121 context = result_contexts.get(target.study_id)
122 recalculated_series = _cash_flow_series_from_context(
123 target=target,
124 context=context,
125 target_label=target_labels[target.study_id],
126 )
127 if recalculated_series:
128 series.extend(recalculated_series)
129 source_row_keys.extend(
130 f"cash_flow.year_{row.year}"
131 for row in (context.discounted_cash_flow if context is not None else ())
132 )
133 continue
134 dataset = _source_dataset(
135 context=context,
136 datasets_by_run=datasets_by_run,
137 chart_key=CHART_CASH_FLOW_NPV,
138 )
139 if dataset is None:
140 continue
141 source_row_keys.extend(_source_row_keys(dataset))
142 warning_refs.extend(_warning_refs(dataset))
143 for source_series in _series(dataset):
144 metric_key = str(source_series.get("key", ""))
145 if metric_key not in _CASH_FLOW_METRIC_LABELS: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 continue
147 points = [
148 _comparison_point(
149 point,
150 target=target,
151 metric_key=metric_key,
152 metric_label=_CASH_FLOW_METRIC_LABELS[metric_key],
153 source_chart_key=CHART_CASH_FLOW_NPV,
154 )
155 for point in _points(source_series)
156 ]
157 if not points: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 continue
159 metric_label = _CASH_FLOW_METRIC_LABELS[metric_key]
160 series.append(
161 {
162 "key": f"{metric_key}.{target.study_id}",
163 "label": f"{target_labels[target.study_id]} - {metric_label}",
164 "unit": str(source_series.get("unit", "")),
165 "points": points,
166 }
167 )
168 if not series:
169 return None
170 return ComparisonChartDatasetPayload(
171 id=-1,
172 chart_key=COMPARISON_CHART_CASH_FLOW,
173 title="Cash Flow Overlay",
174 chart_type="multi_line",
175 source_row_keys=_unique(source_row_keys),
176 chart_data={"series": series},
177 rendering_metadata={
178 "chart_family": COMPARISON_CHART_CASH_FLOW,
179 "x_axis": "project_year",
180 "metric_options": [
181 {"value": key, "label": label}
182 for key, label in _CASH_FLOW_METRIC_LABELS.items()
183 ],
184 "default_metric_keys": ["cumulative_discounted_cash_flow"],
185 "warning_refs": _unique_warning_refs(warning_refs),
186 },
187 )
190def _cash_flow_series_from_context(
191 *,
192 target: ComparisonTarget,
193 context: ResultContext | None,
194 target_label: str,
195) -> list[dict[str, JsonValue]]:
196 """Build comparison cash-flow series from transient comparison recalculation rows."""
197 if context is None or not context.discounted_cash_flow:
198 return []
200 return [
201 {
202 "key": f"{metric_key}.{target.study_id}",
203 "label": f"{target_label} - {metric_label}",
204 "unit": context.run.result_currency,
205 "points": [
206 {
207 "key": f"{target.study_id}.{metric_key}.cash_flow.year_{row.year}",
208 "label": f"Year {row.year}",
209 "value": _cash_flow_metric_value(row, metric_key),
210 "unit": context.run.result_currency,
211 "source_row": {
212 "id": 0,
213 "row_key": f"cash_flow.year_{row.year}",
214 "label": f"Year {row.year}",
215 },
216 "assumptions": [],
217 "warning_refs": [],
218 "metadata": {
219 "point_type": "cash_flow",
220 "year": row.year,
221 "present_value": str(row.present_value),
222 "study_id": target.study_id,
223 "study_name": target.name,
224 "flowsheet_name": target.flowsheet_name,
225 "metric_key": metric_key,
226 "metric_label": metric_label,
227 "source_chart_key": CHART_CASH_FLOW_NPV,
228 },
229 }
230 for row in context.discounted_cash_flow
231 ],
232 }
233 for metric_key, metric_label in _CASH_FLOW_METRIC_LABELS.items()
234 ]
237def _cash_flow_metric_value(row, metric_key: str) -> str:
238 """Return a JSON-safe cash-flow chart value from a discounted cash-flow row."""
239 if metric_key == "annual_net_cash_flow":
240 return str(row.cash_flow)
241 return str(row.cumulative_present_value)
244def _cost_breakdown_dataset(
245 *,
246 studies: list[ComparisonTarget],
247 result_contexts: dict[int | None, ResultContext | None],
248 datasets_by_run: dict[int, dict[str, EconomicsChartDataset]],
249) -> ComparisonChartDatasetPayload | None:
250 series = []
251 source_row_keys: list[str] = []
252 warning_refs: list[dict[str, JsonValue]] = []
253 target_labels = _target_labels(studies)
254 for target in studies:
255 context = result_contexts.get(target.study_id)
256 points = []
257 for chart_key, metric_label in _COST_BREAKDOWN_METRIC_LABELS.items():
258 dataset = _source_dataset(
259 context=context,
260 datasets_by_run=datasets_by_run,
261 chart_key=chart_key,
262 )
263 if dataset is None:
264 continue
265 source_row_keys.extend(_source_row_keys(dataset))
266 warning_refs.extend(_warning_refs(dataset))
267 for source_series in _series(dataset):
268 for point in _points(source_series):
269 copied = _comparison_point(
270 point,
271 target=target,
272 metric_key=chart_key,
273 metric_label=metric_label,
274 source_chart_key=chart_key,
275 )
276 metadata = _metadata(copied)
277 label = str(copied.get("label", ""))
278 metadata["category_key"] = str(
279 copied.get("source_row", {}).get("row_key")
280 if isinstance(copied.get("source_row"), dict)
281 else copied.get("key", label)
282 )
283 metadata["category_label"] = f"{metric_label}: {label}"
284 copied["metadata"] = metadata
285 points.append(copied)
286 if points:
287 series.append(
288 {
289 "key": str(target.study_id),
290 "label": target_labels[target.study_id],
291 "unit": "mixed",
292 "points": points,
293 }
294 )
295 if not series:
296 return None
297 return ComparisonChartDatasetPayload(
298 id=-2,
299 chart_key=COMPARISON_CHART_COST_BREAKDOWN,
300 title="Cost Breakdown",
301 chart_type="grouped_bar",
302 source_row_keys=_unique(source_row_keys),
303 chart_data={"series": series},
304 rendering_metadata={
305 "chart_family": COMPARISON_CHART_COST_BREAKDOWN,
306 "metric_options": [
307 {"value": key, "label": label}
308 for key, label in _COST_BREAKDOWN_METRIC_LABELS.items()
309 ],
310 "default_metric_keys": [
311 CHART_CAPEX_BREAKDOWN,
312 CHART_OPEX_BREAKDOWN,
313 ],
314 "warning_refs": _unique_warning_refs(warning_refs),
315 },
316 )
319def _operating_timeline_overlay_dataset(
320 *,
321 studies: list[ComparisonTarget],
322 result_contexts: dict[int | None, ResultContext | None],
323 datasets_by_run: dict[int, dict[str, EconomicsChartDataset]],
324 source_chart_key: str,
325 comparison_chart_key: str,
326 title: str,
327 dataset_id: int,
328) -> ComparisonChartDatasetPayload | None:
329 """Overlay one operating-cost timeline dataset for selected studies."""
330 series = []
331 source_row_keys: list[str] = []
332 warning_refs: list[dict[str, JsonValue]] = []
333 target_labels = _target_labels(studies)
334 for target in studies:
335 context = result_contexts.get(target.study_id)
336 dataset = _source_dataset(
337 context=context,
338 datasets_by_run=datasets_by_run,
339 chart_key=source_chart_key,
340 )
341 if dataset is None:
342 continue
343 source_row_keys.extend(_source_row_keys(dataset))
344 warning_refs.extend(_warning_refs(dataset))
345 for source_series in _series(dataset): 345 ↛ 334line 345 didn't jump to line 334 because the loop on line 345 didn't complete
346 points = [
347 _comparison_point(
348 point,
349 target=target,
350 metric_key=str(target.study_id),
351 metric_label=target_labels[target.study_id],
352 source_chart_key=source_chart_key,
353 )
354 for point in _points(source_series)
355 ]
356 if points: 356 ↛ 345line 356 didn't jump to line 345 because the condition on line 356 was always true
357 series.append(
358 {
359 "key": str(target.study_id),
360 "label": target_labels[target.study_id],
361 "unit": str(source_series.get("unit", "")),
362 "points": points,
363 }
364 )
365 break
366 if not series:
367 return None
368 return ComparisonChartDatasetPayload(
369 id=dataset_id,
370 chart_key=comparison_chart_key,
371 title=title,
372 chart_type="multi_line",
373 source_row_keys=_unique(source_row_keys),
374 chart_data={"series": series},
375 rendering_metadata={
376 "chart_family": comparison_chart_key,
377 "x_axis": "operating_hours",
378 "metric_options": [
379 {"value": item["key"], "label": item["label"]} for item in series
380 ],
381 "default_metric_keys": [item["key"] for item in series],
382 "warning_refs": _unique_warning_refs(warning_refs),
383 },
384 )
387def _source_dataset(
388 *,
389 context: ResultContext | None,
390 datasets_by_run: dict[int, dict[str, EconomicsChartDataset]],
391 chart_key: str,
392) -> EconomicsChartDataset | None:
393 if context is None:
394 return None
395 return datasets_by_run.get(context.run.pk, {}).get(chart_key)
398def _comparison_point(
399 point: dict[str, Any],
400 *,
401 target: ComparisonTarget,
402 metric_key: str,
403 metric_label: str,
404 source_chart_key: str,
405) -> dict[str, JsonValue]:
406 copied = deepcopy(point)
407 metadata = _metadata(copied)
408 metadata.update(
409 {
410 "study_id": target.study_id,
411 "study_name": target.name,
412 "flowsheet_name": target.flowsheet_name,
413 "metric_key": metric_key,
414 "metric_label": metric_label,
415 "source_chart_key": source_chart_key,
416 }
417 )
418 copied["key"] = f"{target.study_id}.{metric_key}.{copied.get('key', '')}"
419 copied["metadata"] = metadata
420 return copied
423def _target_labels(studies: list[ComparisonTarget]) -> dict[int | None, str]:
424 """Use flowsheet names when repeated study names would make chart legends ambiguous."""
425 names = [target.name for target in studies]
426 duplicate_names = len(set(names)) < len(names)
427 return {
428 target.study_id: (
429 (target.flowsheet_name or target.name) if duplicate_names else target.name
430 )
431 for target in studies
432 }
435def _series(dataset: EconomicsChartDataset) -> list[dict[str, Any]]:
436 chart_data = dataset.chart_data if isinstance(dataset.chart_data, dict) else {}
437 series = chart_data.get("series", [])
438 return [item for item in series if isinstance(item, dict)]
441def _points(series: dict[str, Any]) -> list[dict[str, Any]]:
442 points = series.get("points", [])
443 return [item for item in points if isinstance(item, dict)]
446def _metadata(point: dict[str, Any]) -> dict[str, JsonValue]:
447 metadata = point.get("metadata", {})
448 return dict(metadata) if isinstance(metadata, dict) else {}
451def _source_row_keys(dataset: EconomicsChartDataset) -> list[str]:
452 keys = dataset.source_row_keys if isinstance(dataset.source_row_keys, list) else []
453 return [str(key) for key in keys]
456def _warning_refs(dataset: EconomicsChartDataset) -> list[dict[str, JsonValue]]:
457 metadata = dataset.rendering_metadata if isinstance(dataset.rendering_metadata, dict) else {}
458 warning_refs = metadata.get("warning_refs", [])
459 return [item for item in warning_refs if isinstance(item, dict)]
462def _unique(values: list[str]) -> list[str]:
463 return list(dict.fromkeys(values))
466def _unique_warning_refs(
467 warning_refs: list[dict[str, JsonValue]],
468) -> list[dict[str, JsonValue]]:
469 keyed = {
470 (
471 str(warning.get("code", "")),
472 str(warning.get("message", "")),
473 str(warning.get("source_row_key", "")),
474 ): warning
475 for warning in warning_refs
476 }
477 return list(keyed.values())