Coverage for src/rok4/Layer.py: 85%

1"""Provide classes to use a layer. 

2 

3The module contains the following classe: 

4 

5- `Layer` - Descriptor to broadcast pyramids' data 

6""" 

7 

8from typing import Dict, List, Tuple, Union 

9import json 

10from json.decoder import JSONDecodeError 

11import os 

12import re 

13 

14from rok4.Exceptions import * 

15from rok4.Pyramid import Pyramid, PyramidType 

16from rok4.TileMatrixSet import TileMatrixSet 

17from rok4.Storage import * 

18from rok4.Utils import * 

19 

20 

21class Layer: 

22 """A data layer, raster or vector 

23 

24 Attributes: 

25 __name (str): layer's technical name 

26 __pyramids (Dict[str, Union[rok4.Pyramid.Pyramid,str,str]]): used pyramids 

27 __format (str): pyramid's list path 

28 __tms (rok4.TileMatrixSet.TileMatrixSet): Used grid 

29 __keywords (List[str]): Keywords 

30 __levels (Dict[str, rok4.Pyramid.Level]): Used pyramids' levels 

31 __best_level (rok4.Pyramid.Level): Used pyramids best level 

32 __resampling (str): Interpolation to use fot resampling 

33 __bbox (Tuple[float, float, float, float]): data bounding box, TMS coordinates system 

34 __geobbox (Tuple[float, float, float, float]): data bounding box, EPSG:4326 

35 """ 

36 

37 @classmethod 

38 def from_descriptor(cls, descriptor: str) -> "Layer": 

39 """Create a layer from its descriptor 

40 

41 Args: 

42 descriptor (str): layer's descriptor path 

43 

44 Raises: 

45 FormatError: Provided path is not a well formed JSON 

46 MissingAttributeError: Attribute is missing in the content 

47 StorageError: Storage read issue (layer descriptor) 

48 MissingEnvironmentError: Missing object storage informations 

49 

50 Returns: 

51 Layer: a Layer instance 

52 """ 

53 try: 

54 data = json.loads(get_data_str(descriptor)) 

55 

56 except JSONDecodeError as e: 

57 raise FormatError("JSON", descriptor, e) 

58 

59 layer = cls() 

60 

61 storage_type, path, root, base_name = get_infos_from_path(descriptor) 

62 layer.__name = base_name[:-5] # on supprime l'extension.json 

63 

64 try: 

65 # Attributs communs 

66 layer.__title = data["title"] 

67 layer.__abstract = data["abstract"] 

68 layer.__load_pyramids(data["pyramids"]) 

69 

70 # Paramètres optionnels 

71 if "keywords" in data: 

72 for k in data["keywords"]: 

73 layer.__keywords.append(k) 

74 

75 if layer.type == PyramidType.RASTER: 

76 if "resampling" in data: 

77 layer.__resampling = data["resampling"] 

78 

79 if "styles" in data: 

80 layer.__styles = data["styles"] 

81 else: 

82 layer.__styles = ["normal"] 

83 

84 # Les bbox, native et géographique 

85 if "bbox" in data: 

86 layer.__geobbox = ( 

87 data["bbox"]["south"], 

88 data["bbox"]["west"], 

89 data["bbox"]["north"], 

90 data["bbox"]["east"], 

91 ) 

92 layer.__bbox = reproject_bbox(layer.__geobbox, "EPSG:4326", layer.__tms.srs, 5) 

93 # On force l'emprise de la couche, on recalcule donc les tuiles limites correspondantes pour chaque niveau 

94 for l in layer.__levels.values(): 

95 l.set_limits_from_bbox(layer.__bbox) 

96 else: 

97 layer.__bbox = layer.__best_level.bbox 

98 layer.__geobbox = reproject_bbox(layer.__bbox, layer.__tms.srs, "EPSG:4326", 5) 

99 

100 except KeyError as e: 

101 raise MissingAttributeError(descriptor, e) 

102 

103 return layer 

104 

105 @classmethod 

106 def from_parameters(cls, pyramids: List[Dict[str, str]], name: str, **kwargs) -> "Layer": 

107 """Create a default layer from parameters 

108 

109 Args: 

110 pyramids (List[Dict[str, str]]): pyramids to use and extrem levels, bottom and top 

111 name (str): layer's technical name 

112 **title (str): Layer's title (will be equal to name if not provided) 

113 **abstract (str): Layer's abstract (will be equal to name if not provided) 

114 **styles (List[str]): Styles identifier to authorized for the layer 

115 **resampling (str): Interpolation to use for resampling 

116 

117 Raises: 

118 Exception: name contains forbidden characters or used pyramids do not shared same parameters (format, tms...) 

119 

120 Returns: 

121 Layer: a Layer instance 

122 """ 

123 

124 layer = cls() 

125 

126 # Informations obligatoires 

127 if not re.match("^[A-Za-z0-9_-]*$", name): 

128 raise Exception( 

129 f"Layer's name have to contain only letters, number, hyphen and underscore, to be URL and storage compliant ({name})" 

130 ) 

131 

132 layer.__name = name 

133 layer.__load_pyramids(pyramids) 

134 

135 # Les bbox, native et géographique 

136 layer.__bbox = layer.__best_level.bbox 

137 layer.__geobbox = reproject_bbox(layer.__bbox, layer.__tms.srs, "EPSG:4326", 5) 

138 

139 # Informations calculées 

140 layer.__keywords.append(layer.type.name) 

141 layer.__keywords.append(layer.__name) 

142 

143 # Informations optionnelles 

144 if "title" in kwargs and kwargs["title"] is not None: 

145 layer.__title = kwargs["title"] 

146 else: 

147 layer.__title = name 

148 

149 if "abstract" in kwargs and kwargs["abstract"] is not None: 

150 layer.__abstract = kwargs["abstract"] 

151 else: 

152 layer.__abstract = name 

153 

154 if layer.type == PyramidType.RASTER: 

155 if "styles" in kwargs and kwargs["styles"] is not None and len(kwargs["styles"]) > 0: 

156 layer.__styles = kwargs["styles"] 

157 else: 

158 layer.__styles = ["normal"] 

159 

160 if "resampling" in kwargs and kwargs["resampling"] is not None: 

161 layer.__resampling = kwargs["resampling"] 

162 

163 return layer 

164 

165 def __init__(self) -> None: 

166 self.__format = None 

167 self.__tms = None 

168 self.__best_level = None 

169 self.__levels = {} 

170 self.__keywords = [] 

171 self.__pyramids = [] 

172 

173 def __load_pyramids(self, pyramids: List[Dict[str, str]]) -> None: 

174 """Load and check pyramids 

175 

176 Args: 

177 pyramids (List[Dict[str, str]]): List of descriptors' paths and optionnaly top and bottom levels 

178 

179 Raises: 

180 Exception: Pyramids' do not all own the same format 

181 Exception: Pyramids' do not all own the same TMS 

182 Exception: Pyramids' do not all own the same channels number 

183 Exception: Overlapping in usage pyramids' levels 

184 """ 

185 

186 ## Toutes les pyramides doivent avoir les même caractéristiques 

187 channels = None 

188 for p in pyramids: 

189 pyramid = Pyramid.from_descriptor(p["path"]) 

190 bottom_level = p.get("bottom_level", None) 

191 top_level = p.get("top_level", None) 

192 

193 if bottom_level is None: 

194 bottom_level = pyramid.bottom_level.id 

195 

196 if top_level is None: 

197 top_level = pyramid.top_level.id 

198 

199 if self.__format is not None and self.__format != pyramid.format: 

200 raise Exception( 

201 f"Used pyramids have to own the same format : {self.__format} != {pyramid.format}" 

202 ) 

203 else: 

204 self.__format = pyramid.format 

205 

206 if self.__tms is not None and self.__tms.id != pyramid.tms.id: 

207 raise Exception( 

208 f"Used pyramids have to use the same TMS : {self.__tms.id} != {pyramid.tms.id}" 

209 ) 

210 else: 

211 self.__tms = pyramid.tms 

212 

213 if self.type == PyramidType.RASTER: 

214 if channels is not None and channels != pyramid.raster_specifications["channels"]: 

215 raise Exception( 

216 f"Used RASTER pyramids have to own the same number of channels : {channels} != {pyramid.raster_specifications['channels']}" 

217 ) 

218 else: 

219 channels = pyramid.raster_specifications["channels"] 

220 self.__resampling = pyramid.raster_specifications["interpolation"] 

221 

222 levels = pyramid.get_levels(bottom_level, top_level) 

223 for l in levels: 

224 if l.id in self.__levels: 

225 raise Exception(f"Level {l.id} is present in two used pyramids") 

226 self.__levels[l.id] = l 

227 

228 self.__pyramids.append( 

229 {"pyramid": pyramid, "bottom_level": bottom_level, "top_level": top_level} 

230 ) 

231 

232 self.__best_level = sorted(self.__levels.values(), key=lambda l: l.resolution)[0] 

233 

234 def __str__(self) -> str: 

235 return f"{self.type.name} layer '{self.__name}'" 

236 

237 @property 

238 def serializable(self) -> Dict: 

239 """Get the dict version of the layer object, descriptor compliant 

240 

241 Returns: 

242 Dict: descriptor structured object description 

243 """ 

244 serialization = { 

245 "title": self.__title, 

246 "abstract": self.__abstract, 

247 "keywords": self.__keywords, 

248 "wmts": {"authorized": True}, 

249 "tms": {"authorized": True}, 

250 "bbox": { 

251 "south": self.__geobbox[0], 

252 "west": self.__geobbox[1], 

253 "north": self.__geobbox[2], 

254 "east": self.__geobbox[3], 

255 }, 

256 "pyramids": [], 

257 } 

258 

259 for p in self.__pyramids: 

260 serialization["pyramids"].append( 

261 { 

262 "bottom_level": p["bottom_level"], 

263 "top_level": p["top_level"], 

264 "path": p["pyramid"].descriptor, 

265 } 

266 ) 

267 

268 if self.type == PyramidType.RASTER: 

269 serialization["wms"] = { 

270 "authorized": True, 

271 "crs": ["CRS:84", "IGNF:WGS84G", "EPSG:3857", "EPSG:4258", "EPSG:4326"], 

272 } 

273 

274 if self.__tms.srs.upper() not in serialization["wms"]["crs"]: 

275 serialization["wms"]["crs"].append(self.__tms.srs.upper()) 

276 

277 serialization["styles"] = self.__styles 

278 serialization["resampling"] = self.__resampling 

279 

280 return serialization 

281 

282 def write_descriptor(self, directory: str = None) -> None: 

283 """Print layer's descriptor as JSON 

284 

285 Args: 

286 directory (str, optional): Directory (file or object) where to print the layer's descriptor, called <layer's name>.json. Defaults to None, JSON is printed to standard output. 

287 """ 

288 content = json.dumps(self.serializable) 

289 

290 if directory is None: 

291 print(content) 

292 else: 

293 put_data_str(content, os.path.join(directory, f"{self.__name}.json")) 

294 

295 @property 

296 def type(self) -> PyramidType: 

297 if self.__format == "TIFF_PBF_MVT": 

298 return PyramidType.VECTOR 

299 else: 

300 return PyramidType.RASTER 

301 

302 @property 

303 def bbox(self) -> Tuple[float, float, float, float]: 

304 return self.__bbox 

305 

306 @property 

307 def geobbox(self) -> Tuple[float, float, float, float]: 

308 return self.__geobbox