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 titles = set() 

70 

71 res = self._query(request) 

72 

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

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

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

76 

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

78 continue 

79 

80 address = ' '.join( 

81 [ 

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

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

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

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

86 ] 

87 ) 

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

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

90 

91 a['title'] = self._get_title(a) 

92 if a['title'] in titles: 

93 continue 

94 

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

96 features.append(self.feature_from_record(rec, mc)) 

97 

98 return features 

99 

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

101 try: 

102 res = gws.lib.net.http_request( 

103 self.serviceUrl, 

104 method='POST', 

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

106 json=request, 

107 ) 

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

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

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

111 return {} 

112 

113 def _get_title(self, a): 

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

115 return f'{a["name"]} ({a["address"]})' 

116 

117 if a['address']: 

118 return a['address'] 

119 

120 subcat = a.get('subcategory') 

121 if not subcat: 

122 return a['name'] 

123 

124 # @TODO 

125 subcat = re.sub(r'^\(\d+\)\s+', '', subcat) 

126 return f'{a["name"]} ({subcat})'