Coverage for gws-app/gws/base/ows/client/parseutil.py: 99%

1"""Parse utilities for OWS capabilities documents.""" 

2 

3from typing import Optional 

4 

5import re 

6 

7import gws 

8import gws.base.metadata 

9import gws.lib.crs 

10import gws.lib.extent 

11import gws.lib.net 

12 

13 

14def service_operations(caps_el: gws.XmlElement) -> list[gws.OwsOperation]: 

15 """Read the operations from a capabilities document. 

16 

17 Supports both the OWS Common format (``OperationsMetadata/Operation``) 

18 and the WMS format (``Capability/Request``). 

19 

20 Args: 

21 caps_el: Root element of the capabilities document. 

22 

23 Returns: 

24 A list of operations, empty if none are found. 

25 """ 

26 

27 # <ows:OperationsMetadata> 

28 # <ows:Operation name="GetCapabilities">... 

29 

30 els = caps_el.findall('OperationsMetadata/Operation') 

31 if els: 

32 return [_parse_operation(e) for e in els] 

33 

34 # <Capability> 

35 # <Request> 

36 # <GetCapabilities>... 

37 

38 el = caps_el.find('Capability/Request') 

39 if el: 

40 return [_parse_operation(e) for e in el] 

41 

42 return [] 

43 

44 

45def _parse_operation(el: gws.XmlElement) -> gws.OwsOperation: 

46 """Read an operation with its allowed parameters, URLs and formats.""" 

47 op = gws.OwsOperation(verb=el.get('name') or el.tag) 

48 

49 # @TODO Range 

50 # @TODO Constraint 

51 

52 # <Parameter name="Format"> 

53 # <AllowedValues> 

54 # <Value>image/gif</Value> 

55 # ... 

56 

57 # <Parameter name="AcceptVersions"> 

58 # <Value>1.0.0</Value> 

59 

60 op.allowedParameters = {} 

61 for param_el in el.findall('Parameter'): 

62 values = param_el.textlist('Value') + param_el.textlist('AllowedValues/Value') 

63 if values: 

64 op.allowedParameters[param_el.get('name').upper()] = values 

65 

66 # <Operation name="GetMap"> 

67 # <DCP> <HTTP> 

68 # <Get xlink:href="...."/> 

69 # <Post xlink:href="..."/> 

70 # </HTTP> </DCP> 

71 # 

72 # 

73 # <GetMap> 

74 # <Format>image/png</Format> 

75 # <DCPType> <HTTP> <Get> 

76 # <OnlineResource xlink:type="simple" xlink:href="..."/> 

77 # </Get> </HTTP> </DCPType> 

78 # </GetMap> 

79 

80 op.postUrl = _parse_url(el.findfirst('DCP/HTTP/Post', 'DCPType/HTTP/Post')) 

81 

82 u = _parse_url(el.findfirst('DCP/HTTP/Get', 'DCPType/HTTP/Get')) 

83 op.url, op.params = gws.lib.net.extract_params(u) 

84 

85 op.mimeTypes = el.textlist('Format') 

86 if 'OUTPUTFORMAT' in op.allowedParameters: 

87 op.mimeTypes.extend(op.allowedParameters['OUTPUTFORMAT']) 

88 

89 return op 

90 

91 

92## 

93 

94 

95def service_metadata(caps_el: gws.XmlElement) -> gws.Metadata: 

96 """Read the service metadata from a capabilities document. 

97 

98 Reads the WMS ``Service`` element or the OWS ``ServiceIdentification`` 

99 and ``ServiceProvider`` elements, including contact information. 

100 

101 Args: 

102 caps_el: Root element of the capabilities document. 

103 

104 Returns: 

105 Normalized metadata. 

106 """ 

107 

108 # wms 

109 # 

110 # <Capabilities 

111 # <Service... 

112 # <Name>... 

113 # <Title>... 

114 # <ContactInformation>... 

115 # 

116 # ows 

117 # 

118 # <Capabilities 

119 # <ows:ServiceIdentification> 

120 # <ows:Title>.... 

121 # <ows:ServiceProvider> 

122 # <ows:ProviderName>... 

123 # <ows:ServiceContact>... 

124 

125 md = gws.base.metadata.new() 

126 

127 _element_metadata(caps_el.findfirst('Service', 'ServiceIdentification'), md) 

128 _contact_metadata(caps_el.findfirst('Service/ContactInformation', 'ServiceProvider/ServiceContact'), md) 

129 

130 md.contactProviderName = caps_el.textof('ServiceProvider/ProviderName') 

131 md.contactProviderSite = caps_el.textof('ServiceProvider/ProviderSite') 

132 

133 # <Capabilities 

134 # <ServiceMetadataURL 

135 

136 service_link = _parse_link(caps_el.find('ServiceMetadataURL')) 

137 if service_link: 

138 md.serviceMetadataURL = service_link.url 

139 

140 return gws.base.metadata.normalize(md) 

141 

142 

143def element_metadata(el: gws.XmlElement) -> gws.Metadata: 

144 """Read the metadata of an element, such as a ``Layer`` or ``FeatureType``. 

145 

146 Args: 

147 el: XML element. 

148 

149 Returns: 

150 Normalized metadata. 

151 """ 

152 

153 # <whatever, e.g. Layer or FeatureType 

154 # <Name... 

155 # <Title... 

156 

157 md = gws.base.metadata.new() 

158 _element_metadata(el, md) 

159 return gws.base.metadata.normalize(md) 

160 

161 

162def _element_metadata(el: Optional[gws.XmlElement], md: gws.Metadata): 

163 """Copy common metadata fields (title, abstract, keywords etc.) from an element into ``md``.""" 

164 if not el: 

165 return 

166 

167 md.abstract = el.textof('Abstract') 

168 md.accessConstraints = el.textof('AccessConstraints') 

169 md.attribution = el.textof('Attribution/Title') 

170 md.fees = el.textof('Fees') 

171 md.keywords = el.textlist('Keywords', 'KeywordList', deep=True) 

172 md.name = el.textof('Name', 'Identifier') 

173 md.title = el.textof('Title') 

174 md.metaLinks = gws.u.compact(_parse_link(e) for e in el.findall('MetadataURL')) 

175 

176 e = el.find('AuthorityURL') 

177 if e: 

178 md.authorityUrl = _parse_url(e) 

179 md.authorityName = e.get('name') 

180 

181 e = el.find('Identifier') 

182 if e: 

183 md.authorityIdentifier = e.text 

184 

185 

186_contact_mapping = [ 

187 # wms 

188 ('contactArea', 'StateOrProvince'), 

189 ('contactCity', 'City'), 

190 ('contactCountry', 'Country'), 

191 ('contactEmail', 'ContactElectronicMailAddress'), 

192 ('contactFax', 'ContactFacsimileTelephone'), 

193 ('contactOrganization', 'ContactOrganization'), 

194 ('contactPerson', 'ContactPerson'), 

195 ('contactPhone', 'ContactVoiceTelephone'), 

196 ('contactPosition', 'ContactPosition'), 

197 ('contactZip', 'PostCode'), 

198 # ows 

199 ('contactArea', 'AdministrativeArea'), 

200 ('contactCity', 'City'), 

201 ('contactCountry', 'Country'), 

202 ('contactEmail', 'ElectronicMailAddress'), 

203 ('contactFax', 'Facsimile'), 

204 ('contactOrganization', 'ProviderName'), 

205 ('contactPerson', 'IndividualName'), 

206 ('contactPhone', 'Voice'), 

207 ('contactPosition', 'PositionName'), 

208 ('contactZip', 'PostalCode'), 

209] 

210 

211 

212def _contact_metadata(el: Optional[gws.XmlElement], md: gws.Metadata): 

213 """Copy contact information from an element into ``md``.""" 

214 if not el: 

215 return 

216 

217 src = el.textdict(deep=True) 

218 

219 for dkey, skey in _contact_mapping: 

220 if skey in src: 

221 setattr(md, dkey, src[skey]) 

222 

223 

224## 

225 

226 

227def wgs_extent(layer_el: gws.XmlElement) -> Optional[gws.Extent]: 

228 """Read the WGS84 bounding box from a ``Layer`` or ``FeatureType`` element. 

229 

230 Extracts coordinates from ``EX_GeographicBoundingBox`` (WMS), ``WGS84BoundingBox`` (OWS) 

231 or ``LatLonBoundingBox``. For the latter, assume x=longitude, y=latitude, 

232 as per OGC 01-068r3, 6.5.6. 

233 

234 Args: 

235 layer_el: ``Layer`` or ``FeatureType`` element. 

236 

237 Returns: 

238 The extent, or ``None`` if the element has no WGS84 bounding box. 

239 """ 

240 

241 el = layer_el.findfirst('EX_GeographicBoundingBox', 'WGS84BoundingBox', 'LatLonBoundingBox') 

242 if el: 

243 bbox = _parse_bbox(el) 

244 if bbox: 

245 return gws.lib.extent.from_list(bbox) 

246 

247 

248def supported_crs(layer_el: gws.XmlElement, extra_crs_ids: list[str] = None) -> list[gws.Crs]: 

249 """Enumerate the supported CRSs of a ``Layer`` or ``FeatureType`` element. 

250 

251 For WMS, enumerates ``CRS``/``SRS`` and ``BoundingBox`` tags, 

252 for OWS, ``DefaultCRS`` and ``OtherCRS``. Unknown CRS ids are skipped. 

253 

254 Args: 

255 layer_el: ``Layer`` or ``FeatureType`` element. 

256 extra_crs_ids: Additional CRS ids. 

257 

258 Returns: 

259 A list of ``Crs`` objects. 

260 """ 

261 

262 # <Layer... 

263 # <CRS>EPSG.... 

264 # <BoundingBox CRS="EPSG:" minx=.... 

265 # 

266 # <FeatureType... 

267 # <DefaultCRS>urn:ogc:def:crs:EPSG... 

268 # <OtherCRS>urn:ogc:def:crs:EPSG... 

269 

270 crsids = set() 

271 

272 for el in layer_el.findall('BoundingBox'): 

273 crsids.add(el.get('SRS') or el.get('CRS')) 

274 

275 for tag in 'DefaultSRS', 'DefaultCRS', 'OtherSRS', 'OtherCRS', 'SRS', 'CRS': 

276 for el in layer_el.findall(tag): 

277 if el.text: 

278 crsids.add(el.text) 

279 

280 crsids.update(extra_crs_ids or []) 

281 

282 return [gws.lib.crs.require(s) for s in crsids if gws.lib.crs.is_valid(s)] 

283 

284 

285## 

286 

287 

288def parse_style(el: gws.XmlElement) -> gws.SourceStyle: 

289 """Read a ``Style`` element. 

290 

291 The style is marked as default if it has ``IsDefault="true"`` 

292 or its name is ``default`` or ends with ``:default``. 

293 

294 Args: 

295 el: ``Style`` element. 

296 

297 Returns: 

298 The source style, with a lower-cased name. 

299 """ 

300 

301 # <Style> 

302 # <Name>default... 

303 # <Title>... 

304 # <LegendURL 

305 # <Format>... 

306 # <OnlineResource... 

307 

308 st = gws.SourceStyle() 

309 

310 st.metadata = element_metadata(el) 

311 st.name = st.metadata.get('name', '').lower() 

312 st.legendUrl = _parse_url(el.findfirst('LegendURL')) 

313 st.isDefault = el.get('IsDefault') == 'true' or st.name == 'default' or st.name.endswith(':default') 

314 return st 

315 

316 

317def default_style(styles: list[gws.SourceStyle]) -> Optional[gws.SourceStyle]: 

318 """Find the default style in a list of styles. 

319 

320 Args: 

321 styles: Source styles. 

322 

323 Returns: 

324 The first style marked as default, otherwise the first style, or ``None`` if the list is empty. 

325 """ 

326 

327 for s in styles: 

328 if s.isDefault: 

329 return s 

330 return styles[0] if styles else None 

331 

332 

333## 

334 

335 

336def to_float(s, default=0.0): 

337 """Convert a string to a float. 

338 

339 Args: 

340 s: String value. 

341 default: Value used when ``s`` is empty. 

342 

343 Returns: 

344 The float value. 

345 

346 Raises: 

347 ``ValueError``: If the string is not a number. 

348 """ 

349 

350 return float(s or default) 

351 

352 

353def to_int(s, default=0): 

354 """Convert a string to an int, accepting float strings. 

355 

356 Args: 

357 s: String value. 

358 default: Value used when ``s`` is empty. 

359 

360 Returns: 

361 The value, truncated to an int. 

362 

363 Raises: 

364 ``ValueError``: If the string is not a number. 

365 """ 

366 

367 # accept floats as well, but convert to int 

368 return int(float(s or default)) 

369 

370 

371def to_float_pair(s): 

372 """Convert a whitespace-separated string to a pair of floats. 

373 

374 Args: 

375 s: String like ``"1.5 2.5"``. 

376 

377 Returns: 

378 The first two numbers as a tuple. 

379 

380 Raises: 

381 ``ValueError``: If a value is not a number. 

382 ``IndexError``: If there are fewer than two values. 

383 """ 

384 

385 s = s.split() 

386 return float(s[0]), float(s[1]) 

387 

388 

389## 

390 

391 

392def _parse_bbox(el: gws.XmlElement): 

393 """Read a bounding box element in any of the supported formats as ``[x1, y1, x2, y2]``, or ``None``.""" 

394 

395 # note: bboxes are always converted to (x1, y1, x2, y2) with x1 < x2, y1 < y2 

396 

397 # <BoundingBox/LatLonBoundingBox CRS="..." minx="0" miny="1" maxx="2" maxy="3"/> 

398 

399 if el.get('minx'): 

400 return [ 

401 to_float(el.get('minx')), 

402 to_float(el.get('miny')), 

403 to_float(el.get('maxx')), 

404 to_float(el.get('maxy')), 

405 ] 

406 

407 # <ows:BoundingBox/WGS84BoundingBox 

408 # <ows:LowerCorner> 0 1 

409 # <ows:UpperCorner> 2 3 

410 

411 if el.findfirst('LowerCorner'): 

412 x1, y1 = to_float_pair(el.textof('LowerCorner')) 

413 x2, y2 = to_float_pair(el.textof('UpperCorner')) 

414 return [ 

415 min(x1, x2), 

416 min(y1, y2), 

417 max(x1, x2), 

418 max(y1, y2), 

419 ] 

420 

421 # <EX_GeographicBoundingBox> 

422 # <westBoundLongitude> 0 

423 # <eastBoundLongitude> 2 

424 # <southBoundLatitude> 1 

425 # <northBoundLatitude> 3 

426 

427 if el.findfirst('westBoundLongitude'): 

428 x1 = to_float(el.textof('eastBoundLongitude')) 

429 y1 = to_float(el.textof('southBoundLatitude')) 

430 x2 = to_float(el.textof('westBoundLongitude')) 

431 y2 = to_float(el.textof('northBoundLatitude')) 

432 return [ 

433 min(x1, x2), 

434 min(y1, y2), 

435 max(x1, x2), 

436 max(y1, y2), 

437 ] 

438 

439 

440def _parse_url(el: gws.XmlElement) -> str: 

441 """Read a URL from an ``href`` attribute or a nested ``OnlineResource``.""" 

442 

443 def cleanup(s): 

444 return (s or '').strip(' ?&') 

445 

446 if not el: 

447 return '' 

448 

449 # <ows:DCP> 

450 # <ows:HTTP> 

451 # <ows:Get xlink:href=... <-- we are here 

452 

453 s = el.get('href') or el.get('onlineResource') 

454 if s: 

455 return cleanup(s) 

456 

457 # <whatever <-- 

458 # <OnlineResource xlink:href=... 

459 

460 e = el.findfirst('OnlineResource') 

461 if e: 

462 return cleanup(e.get('href', default=e.text)) 

463 

464 return '' 

465 

466 

467def _parse_link(el: gws.XmlElement) -> Optional[gws.MetadataLink]: 

468 """Read a metadata link in the simple or nested format.""" 

469 

470 if not el: 

471 return None 

472 

473 # see base/ows/server/templatelib.py 

474 # regarding different MetadataURL formats 

475 

476 # simple 

477 if el.get('href'): 

478 return gws.MetadataLink(url=el.get('href')) 

479 

480 # nested 

481 d = gws.u.strip( 

482 { 

483 'url': _parse_url(el), 

484 'type': el.get('type'), 

485 'format': el.textof('Format'), 

486 } 

487 ) 

488 

489 if d: 

490 return gws.MetadataLink(d)