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.lib.shape 

9import gws.lib.net 

10import gws.lib.jsonx 

11import gws.gis.source 

12import gws.lib.crs 

13import gws.lib.bounds 

14 

15 

16@gws.ext.config.model('nominatim') 

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

18 """Data model for places and addresses from Nominatim.""" 

19 

20 country: Optional[str] 

21 """Countries to limit the results to.""" 

22 language: Optional[str] 

23 """Preferred language of the results.""" 

24 

25 

26@gws.ext.object.model('nominatim') 

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

28 """Nominatim model object. 

29 

30 A read-only model whose features are results of the Nominatim search API. 

31 """ 

32 

33 country: str 

34 """Countries to limit the results to.""" 

35 language: str 

36 """Preferred language of the results.""" 

37 

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

39 

40 def configure(self): 

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

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

43 self.uidName = 'uid' 

44 self.geometryName = 'geometry' 

45 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

46 

47 def props(self, user): 

48 return gws.u.merge( 

49 super().props(user), 

50 canCreate=False, 

51 canDelete=False, 

52 canWrite=False, 

53 ) 

54 

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

56 if not search.shape: 

57 return [] 

58 

59 params = { 

60 'q': search.keyword, 

61 'addressdetails': 1, 

62 'polygon_geojson': 1, 

63 'format': 'json', 

64 'bounded': 1, 

65 'dedupe': 1, 

66 'limit': search.limit, 

67 } 

68 

69 if self.language: 

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

71 

72 if self.country: 

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

74 

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

76 

77 features = [] 

78 

79 for rec in self._query(params): 

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

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

82 

83 if not geom: 

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

85 continue 

86 

87 shape = gws.lib.shape.from_geojson(geom, gws.lib.crs.WGS84, always_xy=True).transformed_to(search.shape.crs) 

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

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

90 continue 

91 

92 features.append(self.feature_from_record( 

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

94 mc)) 

95 

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

97 

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

99 """Send a search request to the service and return the results, or an empty list on error.""" 

100 try: 

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

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

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

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

105 return [] 

106 

107 def _normalize(self, rec): 

108 """Flatten the address, add name and OSM class/type attributes, and drop coordinate attributes.""" 

109 # merge the address subrec into the main 

110 

111 if 'address' in rec: 

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

113 rec['address_' + k] = v 

114 

115 # ensure basic address fields 

116 

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

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

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

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

121 

122 # find out the "name" 

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

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

125 

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

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

128 

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

130 

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

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

133 

134 # remove geographic attributes 

135 

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

137 rec.pop(k, None) 

138 

139 return rec