Coverage for gws-app/gws/base/ows/client/featureinfo.py: 94%

1"""Parser for WMS/WFS feature info responses.""" 

2 

3import gws 

4import gws.lib.shape 

5import gws.lib.gml 

6import gws.lib.xmlx as xmlx 

7 

8 

9class Error(gws.Error): 

10 """Feature info parse error.""" 

11 

12 pass 

13 

14 

15def parse(text: str, default_crs: gws.Crs = None, always_xy=False) -> list[gws.FeatureRecord]: 

16 """Parse a GetFeatureInfo or GetFeature response into feature records. 

17 

18 The format is detected from the name of the XML root element. 

19 An empty response yields an empty list. 

20 

21 Args: 

22 text: Response text. 

23 default_crs: CRS for geometries that do not specify one. 

24 always_xy: If true, coordinates are read in XY order regardless of the CRS axis order. 

25 

26 Returns: 

27 A list of feature records. 

28 

29 Raises: 

30 ``Error``: If the response is not valid XML or the format is not recognized. 

31 """ 

32 

33 gws.debug.time_start('featureinfo:parse') 

34 recs = _parse(text.strip(), default_crs, always_xy) 

35 gws.debug.time_end() 

36 return recs 

37 

38 

39def _parse(text, default_crs, always_xy): 

40 """Detect the response format and run the matching parser.""" 

41 if not text.strip(): 

42 return [] 

43 

44 if text.startswith('<'): 

45 try: 

46 xml_el = xmlx.from_string(text, gws.XmlOptions(removeNamespaces=True)) 

47 except xmlx.Error as exc: 

48 raise Error(f'XML error') from exc 

49 

50 parser = _XML_FORMATS.get(xml_el.name.lower()) 

51 if not parser: 

52 raise Error(f'XML format error for {xml_el.name!r}') 

53 

54 recs = parser(xml_el, default_crs, always_xy) 

55 gws.log.debug(f'parsed with {parser.__name__} count={len(recs)}') 

56 return recs 

57 

58 raise Error(f'unknown format in {text[:100]!r}') 

59 

60 

61## 

62 

63 

64def _parse_msgmloutput(xml_el: gws.XmlElement, default_crs, always_xy): 

65 # msGMLOutput (MapServer) 

66 # 

67 # <msGMLOutput 

68 # <LAYER_1> 

69 # <gml:name>LAYER_NAME 

70 # <FEATURE_1> 

71 # <gml:boundedBy> 

72 # ... 

73 # </gml:boundedBy> 

74 # <GEOMETRY> 

75 # <gml:Point... 

76 # </GEOMETRY> 

77 # <attr>....</attr> 

78 # <attr>....</attr> 

79 # 

80 

81 recs = [] 

82 

83 for layer_el in xml_el: 

84 layer_name = layer_el.name.lower() 

85 for el in layer_el: 

86 if el.isa('name'): 

87 layer_name = el.text 

88 else: 

89 rec = _record_from_gml(el, default_crs, always_xy) 

90 rec.meta = {'layerName': layer_name} 

91 recs.append(rec) 

92 

93 return recs 

94 

95 

96def _parse_featurecollection(xml_el: gws.XmlElement, default_crs, always_xy): 

97 # FeatureCollection (OGC) 

98 # 

99 # <FeatureCollection 

100 # <wfs:member> 

101 # <FEATURE gml:id=... 

102 # <attr>....</attr> 

103 # <attr> <nested>....</attr> 

104 # <GEOMETRY> 

105 # <gml:Point... 

106 # 

107 

108 recs = [] 

109 

110 for member_el in xml_el: 

111 if member_el.isa('member', 'featureMember'): 

112 if len(member_el) == 1: 

113 # <wfs:member><my:feature><attr... 

114 recs.append(_record_from_gml(member_el[0], default_crs, always_xy)) 

115 elif len(member_el) > 1: 

116 # <wfs:member><attr... 

117 recs.append(_record_from_gml(member_el, default_crs, always_xy)) 

118 elif member_el.isa('featureMembers'): 

119 # WFS 1.1.0: features directly inside <gml:featureMembers> 

120 for feature_el in member_el: 

121 recs.append(_record_from_gml(feature_el, default_crs, always_xy)) 

122 

123 return recs 

124 

125 

126def _parse_getfeatureinforesponse(xml_el: gws.XmlElement, default_crs, always_xy): 

127 # GetFeatureInfoResponse (geoserver/qgis) 

128 # 

129 # <GetFeatureInfoResponse> 

130 # <Layer name="...."> 

131 # <Feature id="..."> 

132 # <Attribute name="..." value="..."/> 

133 # <Attribute name="geometry" value="<wkt>"/> 

134 # 

135 # For qgis raster layers, "Attribute" is directly under "Layer": 

136 # 

137 # <GetFeatureInfoResponse> 

138 # <Layer name="...."> 

139 # <Attribute name="..." value="..."/> 

140 

141 def attr(rec, el): 

142 key = el.get('name').lower() 

143 val = el.get('value', '').strip() 

144 if key == 'geometry': 

145 rec.shape = gws.lib.shape.from_wkt(val, default_crs) 

146 elif len(val) > 0: 

147 rec.attributes[key] = val 

148 

149 recs = [] 

150 

151 for layer_el in xml_el: 

152 layer_name = layer_el.get('name') 

153 

154 raster_rec = gws.FeatureRecord( 

155 attributes={}, 

156 uid='', 

157 meta={'layerName': layer_name}, 

158 ) 

159 

160 for sub_el in layer_el: 

161 if sub_el.isa('Feature'): 

162 rec = gws.FeatureRecord( 

163 attributes={}, 

164 uid=_get_uid(sub_el), 

165 meta={'layerName': layer_name}, 

166 ) 

167 for el in sub_el: 

168 if el.isa('Attribute'): 

169 attr(rec, el) 

170 recs.append(rec) 

171 

172 if sub_el.isa('Attribute'): 

173 attr(raster_rec, sub_el) 

174 

175 if raster_rec.attributes: 

176 recs.append(raster_rec) 

177 

178 return recs 

179 

180 

181def _parse_featureinforesponse(xml_el: gws.XmlElement, default_crs, always_xy): 

182 # FeatureInfoResponse (Arcgis) 

183 # 

184 # https://webhelp.esri.com/arcims/9.3/General/mergedProjects/wms_connect/wms_connector/get_featureinfo.htm 

185 # 

186 # <FeatureInfoResponse... 

187 # <fields objectid="15111" shape="polygon"... 

188 # <fields objectid="15111" shape="polygon"... 

189 

190 recs = [] 

191 

192 for fields_el in xml_el: 

193 if fields_el.isa('FIELDS'): 

194 rec = gws.FeatureRecord( 

195 attributes={}, 

196 uid=_get_uid(fields_el), 

197 ) 

198 for key, val in fields_el.attrib.items(): 

199 key = key.lower() 

200 if key != 'shape': 

201 rec.attributes[key] = val 

202 recs.append(rec) 

203 

204 return recs 

205 

206 

207def _parse_geobak(xml_el: gws.XmlElement, default_crs, always_xy): 

208 # GeoBAK (https://www.egovernment.sachsen.de/geodaten.html) 

209 # 

210 # <geobak_20:Sachdatenabfrage... 

211 # <geobak_20:Kartenebene>.... 

212 # <geobak_20:Inhalt> 

213 # <geobak_20:Datensatz> 

214 # <geobak_20:Attribut> 

215 # <geobak_20:Name>... 

216 # <geobak_20:Wert>... 

217 # <geobak_20:Inhalt> 

218 # <geobak_20:Datensatz> 

219 # ... 

220 

221 recs = [] 

222 layer_name = '' 

223 

224 for el in xml_el: 

225 if el.isa('Kartenebene'): 

226 layer_name = el.text 

227 continue 

228 

229 if el.isa('Inhalt'): 

230 rec = gws.FeatureRecord(attributes={}, meta={'layerName': layer_name}) 

231 for attr_el in el[0]: 

232 key = attr_el[0].text.strip().lower() 

233 val = attr_el[1].text.strip() 

234 if key != 'shape' and val.lower() != 'null': 

235 rec.attributes[key] = val 

236 recs.append(rec) 

237 

238 return recs 

239 

240 

241def _parse_osiris(xml_el: gws.XmlElement, default_crs, always_xy): 

242 # OSIRIS (https://www.gis-consult.de/produkte-und-loesungen/gc-osiris/) 

243 # 

244 # <OSIRIS_Objektmenge ... 

245 # <OSIRIS_Objekt fid="..." 

246 # <gml:description>... 

247 # <gml:name>... 

248 # <gml:boundedBy>... 

249 

250 recs = [] 

251 

252 for obj_el in xml_el: 

253 if obj_el.isa('OSIRIS_Objekt'): 

254 rec = _record_from_gml(obj_el, default_crs, always_xy) 

255 recs.append(rec) 

256 

257 return recs 

258 

259 

260## 

261 

262_DEEP_ATTRIBUTE_DELIMITER = '.' 

263 

264 

265def _record_from_gml(feature_el, default_crs, always_xy) -> gws.FeatureRecord: 

266 """Create a record from a GML feature element; nested elements become dotted attribute names.""" 

267 

268 # like GDAL does: 

269 # "When reading a feature, the driver will by default only take into account 

270 # the last recognized GML geometry found..." (https://gdal.org/drivers/vector/gml.html) 

271 

272 rec = gws.FeatureRecord( 

273 attributes={}, 

274 uid=_get_uid(feature_el), 

275 meta={'layerName': feature_el.name.lower()}, 

276 ) 

277 

278 bbox = None 

279 

280 for el in feature_el: 

281 if el.isa('boundedBy'): 

282 # <gml:boundedBy directly under feature 

283 bbox = gws.lib.gml.parse_envelope(el[0], default_crs, always_xy) 

284 elif gws.lib.gml.is_geometry_element(el): 

285 # <gml:Polygon etc directly under feature 

286 rec.shape = gws.lib.gml.parse_shape(el, default_crs, always_xy) 

287 elif len(el) == 1 and gws.lib.gml.is_geometry_element(el[0]): 

288 # <gml:Polygon etc in a wrapper tag 

289 rec.shape = gws.lib.gml.parse_shape(el[0], default_crs, always_xy) 

290 elif len(el) > 0: 

291 # sub-feature 

292 sub = _record_from_gml(el, default_crs, always_xy) 

293 for k, v in sub.attributes.items(): 

294 rec.attributes[el.name.lower() + _DEEP_ATTRIBUTE_DELIMITER + k] = v 

295 else: 

296 # attribute <attr>text</attr> 

297 s = el.text.strip() 

298 if s: 

299 rec.attributes[el.name.lower()] = s 

300 

301 if not rec.shape and bbox: 

302 rec.shape = gws.lib.shape.from_bounds(bbox) 

303 

304 return rec 

305 

306 

307_UIDS = ['id', 'fid', 'objectid', 'ID', 'FID', 'OBJECTID'] 

308 

309 

310def _get_uid(el): 

311 """Return the feature id from the first known id attribute, or an empty string.""" 

312 for u in _UIDS: 

313 if u in el.attrib: 

314 return el.get(u) 

315 return '' 

316 

317 

318## 

319 

320 

321_XML_FORMATS = { 

322 'msgmloutput': _parse_msgmloutput, 

323 'featurecollection': _parse_featurecollection, 

324 'getfeatureinforesponse': _parse_getfeatureinforesponse, 

325 'featureinforesponse': _parse_featureinforesponse, 

326 'sachdatenabfrage': _parse_geobak, 

327 'osiris_objektmenge': _parse_osiris, 

328}