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

1"""Nominatim model.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.base.model 

7import gws.base.feature 

8import gws.base.shape 

9import gws.lib.net 

10import gws.lib.jsonx 

11import gws.gis.source 

12import gws.gis.crs 

13import gws.gis.bounds 

14 

15 

16gws.ext.new.model('nominatim') 

17 

18 

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

20 """Nominatim model""" 

21 

22 country: Optional[str] 

23 """country to limit the search""" 

24 language: Optional[str] 

25 """language to return the results in""" 

26 

27 

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

29 """Nominatim model.""" 

30 

31 country: str 

32 language: str 

33 

34 serviceUrl = 'https://nominatim.openstreetmap.org/search' 

35 

36 def configure(self): 

37 self.country = self.cfg('country') 

38 self.language = self.cfg('language') 

39 self.uidName = 'uid' 

40 self.geometryName = 'geometry' 

41 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

42 

43 def props(self, user): 

44 return gws.u.merge( 

45 super().props(user), 

46 canCreate=False, 

47 canDelete=False, 

48 canWrite=False, 

49 ) 

50 

51 def find_features(self, search, user, **kwargs): 

52 params = { 

53 'q': search.keyword, 

54 'addressdetails': 1, 

55 'polygon_geojson': 1, 

56 'format': 'json', 

57 'bounded': 1, 

58 'dedupe': 1, 

59 'limit': search.limit, 

60 } 

61 

62 if self.language: 

63 params['accept-language'] = self.cfg('language') 

64 

65 if self.country: 

66 params['countrycodes'] = self.cfg('country') 

67 

68 params['viewbox'] = gws.gis.bounds.transform(search.shape.bounds(), gws.gis.crs.WGS84).extent 

69 

70 features = [] 

71 

72 for rec in self._query(params): 

73 uid = rec.get('place_id') or rec.get('osm_id') 

74 geom = rec.pop('geojson', {}) 

75 

76 if not geom: 

77 gws.log.debug(f'SKIP {uid}: no geometry') 

78 continue 

79 

80 shape = gws.base.shape.from_geojson(geom, gws.gis.crs.WGS84, always_xy=True).transformed_to(search.shape.crs) 

81 if not shape.intersects(search.shape): 

82 gws.log.debug(f'SKIP {uid}: no intersection') 

83 continue 

84 

85 features.append(self.feature_from_record( 

86 gws.FeatureRecord(uid=uid, shape=shape, attributes=self._normalize(rec)), 

87 user)) 

88 

89 return sorted(features, key=lambda f: (f.get('name'), f.get('osm_class'), f.get('osm_type'))) 

90 

91 def _query(self, params) -> list[dict]: 

92 try: 

93 res = gws.lib.net.http_request(self.serviceUrl, params=params) 

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

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

96 gws.log.error('nominatim request error', e) 

97 return [] 

98 

99 def _normalize(self, rec): 

100 # merge the address subrec into the main 

101 

102 if 'address' in rec: 

103 for k, v in rec.pop('address').items(): 

104 rec['address_' + k] = v 

105 

106 # ensure basic address fields 

107 

108 rec['address_road'] = rec.get('address_road') or '' 

109 rec['address_building'] = rec.get('address_building') or rec.get('address_house_number') or '' 

110 rec['address_city'] = rec.get('address_city') or rec.get('address_town') or rec.get('address_village') or '' 

111 rec['address_country'] = rec.get('address_country') or '' 

112 

113 # find out the "name" 

114 # the problem is there's no fixed key for the name (the key depends on the class) 

115 # however, display_name seems to always start with the 'name' 

116 

117 dn = rec.get('display_name', '') 

118 rec['name'] = dn.split(',')[0].strip() 

119 

120 # rename 'class' and 'type' for easier templating 

121 

122 rec['osm_class'] = rec.pop('class', '') 

123 rec['osm_type'] = rec.pop('type', '') 

124 

125 # remove geographic attributes 

126 

127 for k in 'boundingbox', 'lat', 'lon': 

128 rec.pop(k, None) 

129 

130 return rec