Coverage for backend/django/core/auxiliary/services/scenario_time_series.py: 100%

1from core.auxiliary.models.Scenario import Scenario, ScenarioTabTypeEnum 

2 

3 

4MSS_TIME_SERIES_FIELDS = ( 

5 "state_name", 

6 "enable_dynamics", 

7 "mss_time_series_enabled", 

8 "mss_time_series_interval", 

9 "mss_time_series_unit", 

10) 

11 

12 

13def is_mss_time_series(scenario: Scenario) -> bool: 

14 """Return whether MSS rows should be interpreted as elapsed-time points.""" 

15 return bool( 

16 scenario.state_name == ScenarioTabTypeEnum.MultiSteadyState 

17 and scenario.mss_time_series_enabled 

18 and not scenario.enable_dynamics 

19 ) 

20 

21 

22def get_mss_time_value(scenario: Scenario, row_index: int) -> int: 

23 """Convert an MSS row index to elapsed time using the scenario interval.""" 

24 if not is_mss_time_series(scenario): 

25 return row_index 

26 return row_index * scenario.mss_time_series_interval 

27 

28 

29def mss_time_series_settings_snapshot(scenario: Scenario) -> dict[str, object]: 

30 """Capture persisted MSS time-series settings before a scenario update.""" 

31 

32 return { 

33 field: getattr(scenario, field) 

34 for field in MSS_TIME_SERIES_FIELDS 

35 } 

36 

37 

38def mss_time_series_settings_changed( 

39 scenario: Scenario, 

40 *, 

41 previous_values: dict[str, object], 

42) -> bool: 

43 """Return whether persisted MSS time-series settings changed.""" 

44 

45 return any( 

46 getattr(scenario, field) != previous_values[field] 

47 for field in MSS_TIME_SERIES_FIELDS 

48 ) 

49 

50 

51def handle_mss_time_series_configuration_changed(scenario: Scenario) -> None: 

52 """Coordinate side effects for production-schedule timing changes.""" 

53 

54 _mark_economics_schedule_configuration_changed(scenario) 

55 _invalidate_result_summary_cache_for_time_series_settings(scenario) 

56 

57 

58def _mark_economics_schedule_configuration_changed(scenario: Scenario) -> None: 

59 from Economics.scheduling.events import schedule_scenario_configuration_changed 

60 

61 schedule_scenario_configuration_changed(scenario_ids=[scenario.id]) 

62 

63 

64def _invalidate_result_summary_cache_for_time_series_settings(scenario: Scenario) -> None: 

65 from core.auxiliary.services.result_summary.cache import ( 

66 ResultSummaryCacheInvalidationReason, 

67 invalidate_result_summary_cache_for_scenario, 

68 ) 

69 

70 invalidate_result_summary_cache_for_scenario( 

71 scenario_id=scenario.id, 

72 reason=ResultSummaryCacheInvalidationReason.mss_time_series_settings_changed, 

73 )