Coverage for src/rok4_tools/global_utils/source.py: 78%

1"""Provide classes to load sources of data. 

2 

3The module contains the following classes: 

4 

5- `Source` - Source objects 

6- `SourcePyramids` - Load pyramids 

7""" 

8 

9from typing import Dict, List, Tuple, Union 

10 

11from rok4.enums import PyramidType 

12from rok4.pyramid import Pyramid 

13 

14 

15class Source: 

16 """Sources to load to create a new pyramid 

17 

18 Attributes: 

19 __bottom (str): Level of the new pyramid's TMS for which the source is used 

20 __top (str): Level of the new pyramid's TMS until which the source is used 

21 """ 

22 

23 def __init__(self, bottom: str, top: str) -> None: 

24 self.__bottom = bottom 

25 self.__top = top 

26 

27 @property 

28 def bottom(self) -> str: 

29 return self.__bottom 

30 

31 @property 

32 def top(self) -> str: 

33 return self.__top 

34 

35 

36class SourcePyramids(Source): 

37 """Pyramid sources to load to create a new pyramid 

38 

39 Attributes: 

40 __tms (TileMatrxSet): Tile Matrix Set of the sources 

41 __format (str): Format of the sources 

42 __pyramids (List[Pyramid]): List of the loaded pyramid sources 

43 __type (PyramidType) : Type of the sources 

44 __channels (int) : If raster pyramid, number of channels of the sources 

45 """ 

46 

47 def __init__(self, bottom: str, top: str, descriptors: List[str]) -> None: 

48 Source.__init__(self, bottom, top) 

49 

50 self.__tms = None 

51 self.__format = None 

52 self.__pyramids = [] 

53 self.__type = None 

54 

55 width_slabs = {} 

56 height_slabs = {} 

57 

58 for d in descriptors: 

59 # Chargement de la pyramide source 

60 try: 

61 pyramid = Pyramid.from_descriptor(d) 

62 self.__pyramids.append(pyramid) 

63 

64 if pyramid.storage_s3_cluster is not None: 

65 # On ne travaille que sur un unique cluster S3, il ne faut pas préciser lequel dans les chemins 

66 raise Exception( 

67 f"Do not set S3 cluster host into bucket name ({d}) : only one cluster can be used for sources" 

68 ) 

69 

70 except Exception as e: 

71 raise Exception(f"Cannot load source pyramid descriptor : {d} : {e}") 

72 

73 # Vérification de l'unicité des caractéristiques des pyramides 

74 if self.__format is None: 

75 self.__tms = pyramid.tms 

76 self.__format = pyramid.format 

77 else: 

78 if self.__tms.name != pyramid.tms.name: 

79 raise Exception( 

80 f"Sources pyramids cannot have two different TMS : {self.__tms.name} and {pyramid.tms.name}" 

81 ) 

82 

83 if self.__format != pyramid.format: 

84 raise Exception( 

85 f"Sources pyramids cannot have two different format : {self.__format} and {pyramid.format}" 

86 ) 

87 

88 # Vérification de l'unicité du type des pyramides sources 

89 if self.__type is None: 

90 self.__type = pyramid.type 

91 if pyramid.type == PyramidType.RASTER: 

92 self.__channels = pyramid.channels 

93 else: 

94 if self.__type != pyramid.type: 

95 raise Exception( 

96 f"Sources pyramids cannot be of two types different : {self.__type} and {self.__type}" 

97 ) 

98 if pyramid.type == PyramidType.RASTER: 

99 if self.__channels != pyramid.channels: 

100 raise Exception( 

101 f"Sources pyramids cannot have two different numbers of channels : {self.__channels} and {pyramid.channels}" 

102 ) 

103 

104 # Vérification de la présence des niveaux 

105 try: 

106 levels = pyramid.get_levels(bottom, top) 

107 except Exception as e: 

108 raise Exception(f"All levels between {bottom} -> {top} are not in {pyramid.name}") 

109 

110 # Vérification de l'unicité de la taille des dalles par niveau 

111 for level in levels: 

112 if level.id in width_slabs: 

113 if ( 

114 width_slabs[level.id] != level.slab_width 

115 or height_slabs[level.id] != level.slab_height 

116 ): 

117 raise Exception( 

118 f"The number of tiles by slab is different between {pyramid.name} and {self.__pyramids[0].name} at level {level.id}" 

119 ) 

120 else: 

121 width_slabs[level.id] = level.slab_width 

122 height_slabs[level.id] = level.slab_height 

123 

124 @property 

125 def tms(self) -> str: 

126 return self.__tms 

127 

128 @property 

129 def format(self) -> str: 

130 return self.__format 

131 

132 @property 

133 def pyramids(self) -> List[Pyramid]: 

134 return self.__pyramids 

135 

136 @property 

137 def type(self) -> PyramidType: 

138 return self.__type 

139 

140 @property 

141 def channels(self) -> int: 

142 """Get the number of channels for RASTER sources 

143 

144 Returns: 

145 int: Number of channels, None if VECTOR sources 

146 """ 

147 return self.__channels 

148 

149 def info_level(self, id_level: str) -> Tuple[int, int, Dict[str, int]]: 

150 """Calculate informations from a level from the level's informations of each pyramid of the datasource 

151 

152 Args: 

153 id_level (str) : name of the level 

154 

155 Returns: 

156 Tuple[int, int, Dict[str, int]] : slab's width of the level, slab's height of the level and terrain extent in TMS coordinates system 

157 """ 

158 slab_width = None 

159 slab_height = None 

160 tile_limits = None 

161 

162 for pyramid in self.__pyramids: 

163 level = pyramid.get_level(id_level) 

164 tile_limits_level = level.tile_limits 

165 if slab_width != None: 

166 if tile_limits_level["min_row"] < tile_limits["min_row"]: 

167 tile_limits["min_row"] = tile_limits_level["min_row"] 

168 if tile_limits_level["min_col"] < tile_limits["min_col"]: 

169 tile_limits["min_col"] = tile_limits_level["min_col"] 

170 if tile_limits_level["max_row"] > tile_limits["max_row"]: 

171 tile_limits["max_row"] = tile_limits_level["max_row"] 

172 if tile_limits_level["max_col"] > tile_limits["max_col"]: 

173 tile_limits["max_col"] = tile_limits_level["max_col"] 

174 else: 

175 slab_width = level.slab_width 

176 slab_height = level.slab_height 

177 tile_limits = tile_limits_level 

178 

179 return (slab_width, slab_height, tile_limits)