Coverage for test/unit/elements/aggregation/test_aggregation.py: 98%

1import unittest 

2from typing import List, Tuple 

3 

4import networkx as nx 

5 

6import bim2sim.elements.aggregation 

7import bim2sim.elements.aggregation.hvac_aggregations 

8from bim2sim.elements.mapping.attribute import Attribute, multi_calc 

9from bim2sim.elements.base_elements import ProductBased 

10from bim2sim.elements import aggregation 

11from bim2sim.elements import hvac_elements as hvac 

12from bim2sim.elements.hvac_elements import HVACPort 

13from bim2sim.elements.graphs.hvac_graph import HvacGraph 

14from test.unit.elements.helper import SetupHelperHVAC 

15 

16 

17class SampleElement(ProductBased): 

18 attr1 = Attribute() 

19 attr2 = Attribute() 

20 attr3 = Attribute() 

21 

22 

23class SampleElementAggregation( 

24 bim2sim.elements.aggregation.AggregationMixin, SampleElement): 

25 

26 def _calc_attr1(self, name): 

27 total = 0 

28 for e in self.elements: 

29 total += e.attr1 

30 return total / len(self.elements) 

31 attr1 = SampleElement.attr1.to_aggregation(_calc_attr1) 

32 

33 @multi_calc 

34 def _calc_multi(self): 

35 return { 

36 'attr2': 2, 

37 'attr3': 3 

38 } 

39 attr2 = SampleElement.attr2.to_aggregation(_calc_multi) 

40 attr3 = SampleElement.attr3.to_aggregation(_calc_multi) 

41 

42 

43class TestAggregation(unittest.TestCase): 

44 """Test abstract Aggregation""" 

45 

46 def test_instantiation(self): 

47 sample1 = SampleElement(attr1=5) 

48 sample2 = SampleElement(attr1=15) 

49 

50 agg = SampleElementAggregation([sample1, sample2]) 

51 

52 self.assertIsInstance(agg, SampleElement) 

53 self.assertEqual(10, agg.attr1) 

54 self.assertEqual(2, agg.attr2) 

55 self.assertEqual(3, agg.attr3) 

56 

57 

58# --- domain Aggregations --- 

59 

60class SampleHVACElement(hvac.HVACProduct): 

61 attr1 = Attribute() 

62 attr2 = Attribute() 

63 attr3 = Attribute() 

64 

65 

66class SampleHVACElementAggregation( 

67 bim2sim.elements.aggregation.hvac_aggregations.HVACAggregationMixin, SampleHVACElement): 

68 

69 @classmethod 

70 def find_matches(cls, graph: HvacGraph 

71 ) -> Tuple[List[nx.Graph], List[dict]]: 

72 return [graph], [{}] 

73 

74 @classmethod 

75 def get_edge_ports(cls, graph) -> List[HVACPort]: 

76 return [] 

77 

78 def _calc_attr1(self, name): 

79 total = 0 

80 for e in self.elements: 

81 total += e.attr1 

82 return total / len(self.elements) 

83 attr1 = SampleHVACElement.attr1.to_aggregation(_calc_attr1) 

84 

85 @multi_calc 

86 def _calc_multi(self): 

87 return { 

88 'attr2': 2, 

89 'attr3': 3 

90 } 

91 attr2 = SampleHVACElement.attr2.to_aggregation(_calc_multi) 

92 attr3 = SampleHVACElement.attr3.to_aggregation(_calc_multi) 

93 

94 

95class AggregationHelper(SetupHelperHVAC): 

96 

97 def get_setup_sample1(self): 

98 """get consumer circuit made of 2 parallel pumps (equal size), 

99 space heater and pipes""" 

100 flags = {} 

101 with self.flag_manager(flags): 

102 # generator circuit 

103 eles = [self.element_generator( 

104 SampleHVACElement, attr1=10 * (i+1)) for i in range(3)] 

105 

106 # connect 

107 self.connect_strait(eles) 

108 

109 flags['connect'] = [eles[0], eles[-1]] 

110 graph = HvacGraph(eles) 

111 return graph, flags 

112 

113 

114class TestHVACAggregation(unittest.TestCase): 

115 helper = None 

116 

117 @classmethod 

118 def setUpClass(cls): 

119 cls.helper = AggregationHelper() 

120 

121 def tearDown(self) -> None: 

122 self.helper.reset() 

123 

124 def test_instantiation(self): 

125 setup, flags = self.helper.get_setup_sample1() 

126 matches, metas = SampleHVACElementAggregation.find_matches(setup) 

127 agg = SampleHVACElementAggregation(setup, matches[0], **metas[0]) 

128 

129 self.assertIsInstance(agg, SampleHVACElement) 

130 self.assertEqual(20, agg.attr1) 

131 self.assertEqual(2, agg.attr2) 

132 self.assertEqual(3, agg.attr3) 

133 

134 

135if __name__ == '__main__': 

136 unittest.main()