Coverage for gws-app/gws/base/ows/server/templatelib.py: 78%

1"""Helper functions for OWS service templates.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.metadata 

7import gws.lib.gml 

8import gws.lib.uom 

9import gws.lib.datetimex as dtx 

10import gws.lib.xmlx as xmlx 

11import gws.base.ows.server as server 

12 

13from . import core, service 

14 

15 

16# OGC 06-121r9 Table 34 

17# Ordered sequence of two double values in decimal degrees, with longitude before latitude 

18def ows_wgs84_bounding_box(lc: core.LayerCaps): 

19 return ( 

20 'ows:WGS84BoundingBox', 

21 ('ows:LowerCorner', coord_dms(lc.layer.wgsExtent[0]), ' ', coord_dms(lc.layer.wgsExtent[1])), 

22 ('ows:UpperCorner', coord_dms(lc.layer.wgsExtent[2]), ' ', coord_dms(lc.layer.wgsExtent[3])), 

23 ) 

24 

25 

26# OGC 06-121r3 sec 7.4.4 

27def ows_service_identification(ta: server.TemplateArgs): 

28 md = ta.service.metadata 

29 

30 return ( 

31 'ows:ServiceIdentification', 

32 ('ows:Title', md.title), 

33 ('ows:Abstract', md.abstract), 

34 ows_keywords(md), 

35 ('ows:ServiceType', ta.service.protocol), 

36 ('ows:ServiceTypeVersion', ta.version), 

37 ('ows:Fees', md.fees) if md.fees else None, 

38 ('ows:AccessConstraints', md.accessConstraints) if md.accessConstraints else None, 

39 ) 

40 

41 

42# OGC 06-121r3 sec 7.4.5 

43def ows_service_provider(ta: server.TemplateArgs): 

44 md = ta.service.metadata 

45 

46 return ( 

47 'ows:ServiceProvider', 

48 ('ows:ProviderName', md.contactProviderName), 

49 ('ows:ProviderSite', {'xlink:href': md.contactProviderSite}), 

50 ( 

51 'ows:ServiceContact', 

52 ('ows:IndividualName', md.contactPerson), 

53 ('ows:PositionName', md.contactPosition), 

54 ( 

55 'ows:ContactInfo', 

56 ('ows:Phone', ('ows:Voice', md.contactPhone), ('ows:Facsimile', md.contactFax)), 

57 ( 

58 'ows:Address', 

59 ('ows:DeliveryPoint', md.contactAddress), 

60 ('ows:City', md.contactCity), 

61 ('ows:AdministrativeArea', md.contactArea), 

62 ('ows:PostalCode', md.contactZip), 

63 ('ows:Country', md.contactCountry), 

64 ('ows:ElectronicMailAddress', md.contactEmail), 

65 ), 

66 ('ows:OnlineResource', {'xlink:href': md.contactUrl}), 

67 ), 

68 ('ows:Role', md.contactRole), 

69 ), 

70 ) 

71 

72 

73# OGC 06-121r3 table 15,16,17 

74# OGC 06-121r3 11.2: 

75# A URL prefix is defined as a string including... mandatory question mark 

76 

77 

78def ows_service_url(ta: server.TemplateArgs, get=True, post=False): 

79 if get: 

80 yield 'ows:DCP/ows:HTTP/ows:Get', {'xlink:type': 'simple', 'xlink:href': ta.serviceUrl + '?'} 

81 if post: 

82 yield 'ows:DCP/ows:HTTP/ows:Post', {'xlink:type': 'simple', 'xlink:href': ta.serviceUrl} 

83 

84 

85def ows_value(value): 

86 return 'ows:Value', value 

87 

88 

89def online_resource(url): 

90 return 'OnlineResource', {'xlink:type': 'simple', 'xlink:href': url} 

91 

92 

93# OGC 01-068r3, 6.2.2 

94# The URL prefix shall end in either a '?' (in the absence of additional server-specific parameters) or a '&'. 

95# OGC 06-042, 6.3.3 

96# A URL prefix is defined... as a string including... mandatory question mark 

97 

98 

99def dcp_service_url(ta: server.TemplateArgs): 

100 return 'DCPType/HTTP/Get', online_resource(ta.serviceUrl + '?') 

101 

102 

103def legend_url_nested(ta: server.TemplateArgs, lc: core.LayerCaps, size=None): 

104 name = xmlx.namespace.unqualify_name(lc.layerNameQ) 

105 return ( 

106 'LegendURL', 

107 {'width': size[0], 'height': size[1]} if size else {}, 

108 ('Format', 'image/png'), 

109 online_resource(f'{ta.serviceUrl}?request=GetLegendGraphic&layer={name}'), 

110 ) 

111 

112 

113def legend_url(ta: server.TemplateArgs, lc: core.LayerCaps, size=None): 

114 name = xmlx.namespace.unqualify_name(lc.layerNameQ) 

115 return ( 

116 'LegendURL', 

117 { 

118 'format': 'image/png', 

119 'xlink:href': f'{ta.serviceUrl}?request=GetLegendGraphic&layer={name}', 

120 }, 

121 ) 

122 

123 

124def ows_keywords(md: gws.Metadata): 

125 return [_ows_keyword_group(kg) for kg in gws.base.metadata.keyword_groups(md)] 

126 

127 

128def _ows_keyword_group(kg: gws.base.metadata.KeywordGroup): 

129 tags = [] 

130 for kw in kg.keywords: 

131 tags.append(('ows:Keyword', kw)) 

132 if kg.codeSpace: 

133 tags.append(('ows:Type', {'codeSpace': kg.codeSpace}, kg.typeName)) 

134 return 'ows:Keywords', tags 

135 

136 

137def wms_keywords(md: gws.Metadata, with_vocabulary: bool = False): 

138 tags = [] 

139 for kg in gws.base.metadata.keyword_groups(md): 

140 for kw in kg.keywords: 

141 if kg.codeSpace and with_vocabulary: 

142 tags.append(('Keyword', kw, {'vocabulary': kg.codeSpace})) 

143 else: 

144 tags.append(('Keyword', kw)) 

145 return 'KeywordList', tags 

146 

147 

148def lon_lat_envelope(lc: core.LayerCaps): 

149 return ( 

150 'lonLatEnvelope', 

151 {'srsName': 'urn:ogc:def:crs:OGC:1.3:CRS84'}, 

152 ('gml:pos', coord_dms(lc.layer.wgsExtent[0]), ' ', coord_dms(lc.layer.wgsExtent[1])), 

153 ('gml:pos', coord_dms(lc.layer.wgsExtent[2]), ' ', coord_dms(lc.layer.wgsExtent[3])), 

154 ) 

155 

156 

157# Nested format (WFS 1, WMS): 

158# OGC 06-042 7.2.4.6.11 

159# The "type" attribute indicates the standard... The enclosed <Format> element... etc 

160 

161 

162def meta_links_nested(ta: server.TemplateArgs, md: gws.Metadata): 

163 if md.metaLinks: 

164 for ml in md.metaLinks: 

165 yield meta_url_nested(ta, ml, 'MetadataURL') 

166 

167 

168def meta_url_nested(ta: server.TemplateArgs, ml: gws.MetadataLink, name: str): 

169 if ml: 

170 yield name, {'type': ml.type}, ('Format', ml.format), online_resource(ta.url_for(ml.url)) 

171 

172 

173# Simple format (WFS 2) 

174# OGC 09-025r1 Table 11 

175# The xlink:href element shall be used to reference any metadata. 

176# The optional about attribute may be used to reference the aspect of the element which includes 

177# this wfs:MetadataURL element that this metadata provides more information about. 

178# (whatever that means) 

179 

180 

181def meta_links_simple(ta: server.TemplateArgs, md: gws.Metadata): 

182 if md.metaLinks: 

183 for ml in md.metaLinks: 

184 yield meta_url_simple(ta, ml, 'MetadataURL') 

185 

186 

187def meta_url_simple(ta: server.TemplateArgs, ml: gws.MetadataLink, name: str): 

188 if ml: 

189 yield name, {'xlink:href': ta.url_for(ml.url), 'about': ml.about} 

190 

191 

192def wfs_feature_collection(ta: server.TemplateArgs): 

193 return ( 

194 'wfs:FeatureCollection', 

195 wfs_feature_collection_attributes(ta), 

196 [ 

197 ( 

198 f'wfs:member/{m.layerCaps.featureNameQ if m.layerCaps else "wfs:feature"}', 

199 {'gml:id': gml_format_uid(ta, m.feature.uid())}, 

200 wfs_feature_collection_member(ta, m), 

201 ) 

202 for m in ta.featureCollection.members 

203 ], 

204 ) 

205 

206 

207def wfs_value_collection(ta: server.TemplateArgs): 

208 return ( 

209 'wfs:ValueCollection', 

210 wfs_feature_collection_attributes(ta), 

211 [('wfs:member', gml_format_value(ta, val)) for val in ta.featureCollection.values], 

212 ) 

213 

214 

215def wfs_feature_collection_attributes(ta): 

216 return { 

217 'timeStamp': ta.featureCollection.timestamp, 

218 'numberMatched': ta.featureCollection.numMatched, 

219 'numberReturned': ta.featureCollection.numReturned, 

220 } 

221 

222 

223def wfs_feature_collection_member(ta: server.TemplateArgs, m: server.FeatureCollectionMember): 

224 geom = None 

225 for name, val in m.feature.attributes.items(): 

226 if m.layerCaps: 

227 name = xmlx.namespace.qualify_name(name, m.layerCaps.xmlNamespace) 

228 if isinstance(val, gws.Shape): 

229 geom = name, gml_format_value(ta, val) 

230 else: 

231 yield name, gml_format_value(ta, val) 

232 # QGIS wants geometry as the last element 

233 if geom: 

234 yield geom 

235 

236 

237def gml_format_uid(ta: server.TemplateArgs, uid): 

238 if not uid: 

239 return '_' 

240 s = str(uid) 

241 if s[0].isdigit(): 

242 return '_' + s 

243 return s 

244 

245 

246def gml_format_value(ta, val): 

247 s, ok = xmlx.util.atom_to_string(val) 

248 if ok: 

249 return s 

250 if isinstance(val, gws.Shape): 

251 # NB Qgis wants inline gml xmlns for adhoc schemas 

252 return gws.lib.gml.shape_to_element( 

253 val, 

254 version=ta.gmlVersion, 

255 always_xy=ta.sr.alwaysXY, 

256 with_inline_xmlns=True, 

257 ) 

258 return str(val) 

259 

260 

261# http://inspire.ec.europa.eu/schemas/common/1.0/network.xsd 

262# Scenario 2: Mandatory (where appropriate) metadata elements not mapped to standard capabilities, 

263# plus mandatory language parameters, 

264# plus OPTIONAL MetadataUrl pointing to an INSPIRE Compliant ISO metadata document 

265 

266 

267def inspire_extended_capabilities(ta: server.TemplateArgs): 

268 md = ta.service.metadata 

269 return [ 

270 ( 

271 'inspire_common:ResourceLocator', 

272 ('inspire_common:URL', ta.serviceUrl), 

273 ('inspire_common:MediaType', 'application/xml'), 

274 ), 

275 ('inspire_common:ResourceType', md.inspireResourceType), 

276 ('inspire_common:TemporalReference/inspire_common:DateOfPublication', iso_date(md.dateCreated)), 

277 ( 

278 'inspire_common:Conformity', 

279 ( 

280 'inspire_common:Specification', 

281 # {'xsi:type': 'inspire_common:citationInspireInteroperabilityRegulation'}, 

282 ( 

283 'inspire_common:Title', 

284 'COMMISSION REGULATION (EU) No 1089/2010 of 23 November 2010 implementing Directive 2007/2/EC of the European Parliament and of the Council as regards interoperability of spatial data sets and services', 

285 ), 

286 ('inspire_common:DateOfPublication', '2010-12-08'), 

287 ('inspire_common:URI', 'OJ:L:2010:323:0011:0102:EN:PDF'), 

288 ( 

289 'inspire_common:ResourceLocator', 

290 ('inspire_common:URL', 'http://eur-lex.europa.eu/LexUriServ/LexUriServ.do?uri=OJ:L:2010:323:0011:0102:EN:PDF'), 

291 ('inspire_common:MediaType', 'application/pdf'), 

292 ), 

293 ), 

294 ('inspire_common:Degree', md.inspireDegreeOfConformity), 

295 ), 

296 ( 

297 'inspire_common:MetadataPointOfContact', 

298 ('inspire_common:OrganisationName', md.contactOrganization), 

299 ('inspire_common:EmailAddress', md.contactEmail), 

300 ), 

301 ('inspire_common:MetadataDate', iso_date(md.dateCreated)), 

302 ('inspire_common:SpatialDataServiceType', md.inspireSpatialDataServiceType), 

303 ('inspire_common:MandatoryKeyword/inspire_common:KeywordValue', md.inspireMandatoryKeyword), 

304 ( 

305 'inspire_common:Keyword', 

306 ( 

307 'inspire_common:OriginatingControlledVocabulary', 

308 ('inspire_common:Title', 'INSPIRE themes'), 

309 ('inspire_common:DateOfPublication', '2008-06-01'), 

310 ), 

311 ('inspire_common:KeywordValue', md.inspireThemeNameEn), 

312 ), 

313 ( 

314 'inspire_common:SupportedLanguages', 

315 ('inspire_common:DefaultLanguage/inspire_common:Language', md.languageBib), 

316 ('inspire_common:SupportedLanguage/inspire_common:Language', md.languageBib), 

317 ), 

318 ('inspire_common:ResponseLanguage/inspire_common:Language', md.languageBib), 

319 ] 

320 

321 

322def coord_dms(n): 

323 prec = gws.lib.uom.DEFAULT_PRECISION[gws.Uom.deg] 

324 return f'{round(n, prec):.{prec}f}' 

325 

326 

327def coord_m(n): 

328 prec = gws.lib.uom.DEFAULT_PRECISION[gws.Uom.m] 

329 return f'{round(n, prec):.{prec}f}' 

330 

331 

332def iso_date(d): 

333 dd = dtx.parse(d) 

334 return dtx.to_iso_date_string(dd) if dd else '' 

335 

336 

337def iso_datetime(d): 

338 dd = dtx.parse(d) 

339 return dtx.to_iso_string(dd, with_tz=':') if dd else '' 

340 

341 

342def namespaces_from_caps(ta: server.TemplateArgs) -> dict[str, gws.XmlNamespace]: 

343 return {lc.xmlNamespace.xmlns: lc.xmlNamespace for lc in ta.layerCapsList if lc.xmlNamespace is not None} 

344 

345 

346def to_xml_response( 

347 ta: server.TemplateArgs, 

348 tag, 

349 namespaces: Optional[dict[str, gws.XmlNamespace]] = None, 

350 default_namespace: Optional[gws.XmlNamespace] = None, 

351) -> gws.ContentResponse: 

352 if ta.sr.isSoap: 

353 tag = ['soap:Envelope', ('soap:Header', ''), ('soap:Body', tag)] 

354 namespaces = namespaces or {} 

355 namespaces['soap'] = xmlx.namespace.require('soap') 

356 

357 el = xmlx.tag(*tag) 

358 opts = gws.XmlOptions( 

359 namespaces=namespaces, 

360 defaultNamespace=default_namespace, 

361 withNamespaceDeclarations=True, 

362 withSchemaLocations=True, 

363 withXmlDeclaration=True, 

364 customXmlns=ta.sr.customXmlns, 

365 ) 

366 

367 return cast(service.Object, ta.sr.service).xml_response(el, opts) 

368 

369 

370def to_xml_response_with_doctype( 

371 ta: server.TemplateArgs, 

372 tag, 

373 doctype: str, 

374) -> gws.ContentResponse: 

375 if ta.sr.isSoap: 

376 tag = ['soap:Envelope', ('soap:Header', ''), ('soap:Body', tag)] 

377 

378 el = xmlx.tag(*tag) 

379 opts = gws.XmlOptions( 

380 doctype=doctype, 

381 withNamespaceDeclarations=False, 

382 withSchemaLocations=False, 

383 withXmlDeclaration=True, 

384 ) 

385 return cast(service.Object, ta.sr.service).xml_response(el, opts)