Coverage for gws-app / gws / plugin / gbd_geoservices / model.py: 0%

1"""GBD Geoservices model.""" 

2 

3import re 

4 

5import gws 

6import gws.base.feature 

7import gws.base.model 

8import gws.base.shape 

9import gws.lib.bounds 

10import gws.lib.crs 

11import gws.gis.source 

12import gws.lib.jsonx 

13import gws.lib.net 

14 

15gws.ext.new.model('gbd_geoservices') 

16 

17 

18class Config(gws.base.model.Config): 

19 """GBD Geoservices model.""" 

20 

21 apiKey: str 

22 """API key for GBD Geoservices.""" 

23 

24 

25class Object(gws.base.model.default_model.Object): 

26 """GBD Geoservices model.""" 

27 

28 apiKey: str 

29 

30 serviceUrl = 'https://geoservices.gbd-consult.de/search' 

31 

32 def configure(self): 

33 self.apiKey = self.cfg('apiKey') 

34 self.uidName = 'uid' 

35 self.geometryName = 'geometry' 

36 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

37 

38 def props(self, user): 

39 return gws.u.merge( 

40 super().props(user), 

41 canCreate=False, 

42 canDelete=False, 

43 canWrite=False, 

44 ) 

45 

46 def find_features(self, search, mc, **kwargs): 

47 request = {'page': 0, 'tags': {}} 

48 

49 if search.shape: 

50 geometry_tolerance = 0.0 

51 

52 if search.tolerance: 

53 n, u = search.tolerance 

54 geometry_tolerance = n * (search.resolution or 1) if u == 'px' else n 

55 

56 search_shape = search.shape.tolerance_polygon(geometry_tolerance) 

57 request['viewbox'] = gws.lib.bounds.wgs_extent(search_shape.bounds()) 

58 

59 kw = search.keyword or '' 

60 if kw: 

61 request['intersect'] = 1 

62 # crude heuristics to check if this is an "address" or a "name" 

63 if re.search(r'\s\d', kw): 

64 request['address'] = kw 

65 else: 

66 request['name'] = kw 

67 

68 features = {} 

69 

70 res = self._query(request) 

71 

72 for f in res['results']['features']: 

73 a = {k.replace(':', '_'): v for k, v in sorted(f['properties'].items())} 

74 shape = gws.base.shape.from_geojson(f['geometry'], gws.lib.crs.WGS84, always_xy=True) 

75 

76 if search.shape and search.shape.type == gws.GeometryType.polygon and not shape.intersects(search.shape): 

77 continue 

78 

79 address = ' '.join( 

80 [ 

81 a.get('addr_street') or '', 

82 a.get('addr_housenumber') or '', 

83 a.get('addr_postcode') or '', 

84 a.get('addr_city') or '', 

85 ] 

86 ) 

87 

88 a['address'] = ' '.join(address.split()) 

89 a['name'] = a.get('name') or '' 

90 

91 if a['name'] and a['address']: 

92 a['title'] = a['name'] + ' (' + a['address'] + ')' 

93 else: 

94 a['title'] = a['name'] or a['address'] 

95 

96 rec = gws.FeatureRecord(uid=f['id'], attributes=a, shape=shape) 

97 features[a['title']] = self.feature_from_record(rec, mc) 

98 

99 return [f for _, f in sorted(features.items())] 

100 

101 def _query(self, request) -> dict: 

102 try: 

103 res = gws.lib.net.http_request( 

104 self.serviceUrl, 

105 method='POST', 

106 headers={'x-api-key': self.apiKey}, 

107 json=request, 

108 ) 

109 return gws.lib.jsonx.from_string(res.text) 

110 except gws.lib.net.Error as e: 

111 gws.log.error('geoservices request error', e) 

112 return {}