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 = { 

48 'page': 0, 

49 'tags': {} 

50 } 

51 

52 if search.shape: 

53 geometry_tolerance = 0.0 

54 

55 if search.tolerance: 

56 n, u = search.tolerance 

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

58 

59 search_shape = search.shape.tolerance_polygon(geometry_tolerance) 

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

61 

62 kw = search.keyword or '' 

63 use_address = False 

64 if kw: 

65 request['intersect'] = 1 

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

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

68 request['address'] = kw 

69 use_address = True 

70 else: 

71 request['name'] = kw 

72 

73 features = {} 

74 

75 res = self._query(request) 

76 fs = res['results']['features'] 

77 

78 for f in fs: 

79 rec = gws.FeatureRecord( 

80 uid=f['id'], 

81 attributes={ 

82 k.replace(':', '_'): v 

83 for k, v in sorted(f['properties'].items()) 

84 }, 

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

86 ) 

87 if search.shape and search.shape.type == gws.GeometryType.polygon and not rec['shape'].intersects(search.shape): 

88 continue 

89 

90 a = rec['attributes'] 

91 

92 if use_address and 'addr_housenumber' not in a: 

93 continue 

94 if not use_address and 'name' not in a: 

95 continue 

96 

97 address = ' '.join([ 

98 a.get('addr_street', ''), 

99 a.get('addr_housenumber', ''), 

100 a.get('addr_postcode', ''), 

101 a.get('addr_city', ''), 

102 ]) 

103 address = ' '.join(address.split()) 

104 

105 if use_address: 

106 a['title'] = address 

107 else: 

108 a['title'] = a.get('name') 

109 if address: 

110 a['title'] += ' (' + address + ')' 

111 

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

113 

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

115 

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

117 try: 

118 res = gws.lib.net.http_request( 

119 self.serviceUrl, 

120 method='POST', 

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

122 json=request 

123 ) 

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

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

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

127 return {}