Coverage for backend/django/core/auxiliary/methods/export_scenario_data.py: 81%

1"""Legacy scenario export helpers. 

2 

3These functions still export the older Solution/PropertyValue data model and are 

4not part of the object-storage CSV import pipeline used by DataColumn/DataRow/DataCell. 

5That separation is intentional for now and round-tripping through CSV import/export 

6should not be assumed here. 

7""" 

8 

9from core.auxiliary.models.Scenario import Scenario 

10from core.auxiliary.models.Flowsheet import Flowsheet 

11from core.auxiliary.models.PropertyValue import PropertyValue 

12from core.auxiliary.models.Solution import Solution 

13 

14 

15def values_per_index(scenario: Scenario): 

16 if not scenario.enable_dynamics: 16 ↛ 19line 16 didn't jump to line 19 because the condition on line 16 was always true

17 return 1 

18 else : 

19 return scenario.num_time_steps 

20 

21# Tested in test_mss.py 

22def export_scenario_data(flowsheet: Flowsheet, scenario: Scenario, properties: list[int] | None = None): 

23 # Create a column for each property value 

24 if properties: 

25 solutions = Solution.objects.filter(scenario=scenario, property__property_id__in=properties).order_by("solve_index") 

26 else: 

27 solutions = Solution.objects.filter(scenario=scenario).order_by("solve_index") 

28 

29 property_values = ( 

30 PropertyValue.objects.filter( 

31 flowsheet_state=flowsheet.current_state, 

32 solutions__in=solutions, 

33 ) 

34 .distinct() 

35 .values( 

36 'id', 

37 'property__displayName', 

38 'property__set__simulationObject__componentName', 

39 'indexedItems__displayName' 

40 ) 

41 ) 

42 

43 # Temp container 

44 grouped_data = {} 

45 # Group indexed items by property value id 

46 for row in property_values: 

47 pv_id = row['id'] 

48 

49 if pv_id not in grouped_data: 

50 grouped_data[pv_id] = { 

51 'uo_name': row['property__set__simulationObject__componentName'], 

52 'prop_name': row['property__displayName'], 

53 'indices': [] 

54 } 

55 

56 if row['indexedItems__displayName']: 

57 grouped_data[pv_id]['indices'].append(row['indexedItems__displayName']) 

58 

59 # Create a list of blanks to fill in missing data 

60 blanks = [None for _ in range(values_per_index(scenario))] 

61 

62 columns = {} 

63 data = {} 

64 

65 for pv_id, info in grouped_data.items(): 

66 base_name = f"{info['uo_name']} - {info['prop_name']}" 

67 index_names = " - ".join(info['indices']) # e.g Phases -> Liquid water, Phases -> Vapor water 

68 column_name = f"{base_name} ({index_names})" if index_names else base_name 

69 

70 columns[pv_id] = column_name 

71 data[column_name] = [] 

72 

73 # Populate the columns 

74 current_solve_index = 0 

75 

76 for solution in solutions: 

77 if solution.solve_index > current_solve_index: 

78 # Fill in blanks for missing solve indices 

79 for _ in range(solution.solve_index - current_solve_index - 1): 79 ↛ 80line 79 didn't jump to line 80 because the loop on line 79 never started

80 for column_name in data.keys(): 

81 data[column_name].extend(blanks) 

82 current_solve_index = solution.solve_index 

83 

84 column_name = columns[solution.property_id] 

85 # Because the values are an array, we flatten them into the data column 

86 data[column_name].extend(solution.values) 

87 

88 return data 

89 

90def collate(data: dict[str,list]): 

91 """ 

92 Collate the data into rows for CSV export 

93 """ 

94 # Collate the data into rows 

95 # Each row is a dict with keys as column names 

96 rows = [] 

97 max_length = max((len(v) for v in data.values()),default=0) 

98 for i in range(max_length): 98 ↛ 99line 98 didn't jump to line 99 because the loop on line 98 never started

99 row = {} 

100 for key, values in data.items(): 

101 row[key] = values[i] if i < len(values) else None 

102 rows.append(row) 

103 return rows 

104