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.formats = el.textlist('Format') 

86 if 'OUTPUTFORMAT' in op.allowedParameters: 

87 op.formats.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 

189 ('contactArea', 'StateOrProvince'), 

190 ('contactCity', 'City'), 

191 ('contactCountry', 'Country'), 

192 ('contactEmail', 'ContactElectronicMailAddress'), 

193 ('contactFax', 'ContactFacsimileTelephone'), 

194 ('contactOrganization', 'ContactOrganization'), 

195 ('contactPerson', 'ContactPerson'), 

196 ('contactPhone', 'ContactVoiceTelephone'), 

197 ('contactPosition', 'ContactPosition'), 

198 ('contactZip', 'PostCode'), 

199 

200 # ows 

201 

202 ('contactArea', 'AdministrativeArea'), 

203 ('contactCity', 'City'), 

204 ('contactCountry', 'Country'), 

205 ('contactEmail', 'ElectronicMailAddress'), 

206 ('contactFax', 'Facsimile'), 

207 ('contactOrganization', 'ProviderName'), 

208 ('contactPerson', 'IndividualName'), 

209 ('contactPhone', 'Voice'), 

210 ('contactPosition', 'PositionName'), 

211 ('contactZip', 'PostalCode'), 

212] 

213 

214 

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

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

217 if not el: 

218 return 

219 

220 src = el.textdict(deep=True) 

221 

222 for dkey, skey in _contact_mapping: 

223 if skey in src: 

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

225 

226 

227## 

228 

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

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

231 

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

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

234 as per OGC 01-068r3, 6.5.6. 

235 

236 Args: 

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

238 

239 Returns: 

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

241 """ 

242 

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

244 if el: 

245 bbox = _parse_bbox(el) 

246 if bbox: 

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

248 

249 

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

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

252 

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

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

255 

256 Args: 

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

258 extra_crs_ids: Additional CRS ids. 

259 

260 Returns: 

261 A list of ``Crs`` objects. 

262 """ 

263 

264 # <Layer... 

265 # <CRS>EPSG.... 

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

267 # 

268 # <FeatureType... 

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

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

271 

272 crsids = set() 

273 

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

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

276 

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

278 for el in layer_el.findall(tag): 

279 if el.text: 

280 crsids.add(el.text) 

281 

282 crsids.update(extra_crs_ids or []) 

283 

284 return gws.u.compact(gws.lib.crs.get(s) for s in crsids) 

285 

286 

287## 

288 

289 

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

291 """Read a ``Style`` element. 

292 

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

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

295 

296 Args: 

297 el: ``Style`` element. 

298 

299 Returns: 

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

301 """ 

302 

303 # <Style> 

304 # <Name>default... 

305 # <Title>... 

306 # <LegendURL 

307 # <Format>... 

308 # <OnlineResource... 

309 

310 st = gws.SourceStyle() 

311 

312 st.metadata = element_metadata(el) 

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

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

315 st.isDefault = ( 

316 el.get('IsDefault') == 'true' 

317 or st.name == 'default' 

318 or st.name.endswith(':default')) 

319 return st 

320 

321 

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

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

324 

325 Args: 

326 styles: Source styles. 

327 

328 Returns: 

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

330 """ 

331 

332 for s in styles: 

333 if s.isDefault: 

334 return s 

335 return styles[0] if styles else None 

336 

337 

338## 

339 

340 

341def to_float(s, default=0.0): 

342 """Convert a string to a float. 

343 

344 Args: 

345 s: String value. 

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

347 

348 Returns: 

349 The float value. 

350 

351 Raises: 

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

353 """ 

354 

355 return float(s or default) 

356 

357 

358def to_int(s, default=0): 

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

360 

361 Args: 

362 s: String value. 

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

364 

365 Returns: 

366 The value, truncated to an int. 

367 

368 Raises: 

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

370 """ 

371 

372 # accept floats as well, but convert to int 

373 return int(float(s or default)) 

374 

375 

376def to_float_pair(s): 

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

378 

379 Args: 

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

381 

382 Returns: 

383 The first two numbers as a tuple. 

384 

385 Raises: 

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

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

388 """ 

389 

390 s = s.split() 

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

392 

393 

394## 

395 

396 

397def _parse_bbox(el: gws.XmlElement): 

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

399 

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

401 

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

403 

404 if el.get('minx'): 

405 return [ 

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

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

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

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

410 ] 

411 

412 # <ows:BoundingBox/WGS84BoundingBox 

413 # <ows:LowerCorner> 0 1 

414 # <ows:UpperCorner> 2 3 

415 

416 if el.findfirst('LowerCorner'): 

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

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

419 return [ 

420 min(x1, x2), 

421 min(y1, y2), 

422 max(x1, x2), 

423 max(y1, y2), 

424 ] 

425 

426 # <EX_GeographicBoundingBox> 

427 # <westBoundLongitude> 0 

428 # <eastBoundLongitude> 2 

429 # <southBoundLatitude> 1 

430 # <northBoundLatitude> 3 

431 

432 if el.findfirst('westBoundLongitude'): 

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

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

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

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

437 return [ 

438 min(x1, x2), 

439 min(y1, y2), 

440 max(x1, x2), 

441 max(y1, y2), 

442 ] 

443 

444 

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

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

447 

448 def cleanup(s): 

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

450 

451 if not el: 

452 return '' 

453 

454 # <ows:DCP> 

455 # <ows:HTTP> 

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

457 

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

459 if s: 

460 return cleanup(s) 

461 

462 # <whatever <-- 

463 # <OnlineResource xlink:href=... 

464 

465 e = el.findfirst('OnlineResource') 

466 if e: 

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

468 

469 return '' 

470 

471 

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

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

474 

475 if not el: 

476 return None 

477 

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

479 # regarding different MetadataURL formats 

480 

481 # simple 

482 if el.get('href'): 

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

484 

485 # nested 

486 d = gws.u.strip({ 

487 'url': _parse_url(el), 

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

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

490 }) 

491 

492 if d: 

493 return gws.MetadataLink(d)