Coverage for gws-app/gws/plugin/geojson/provider.py: 79%

1"""GeoJSON provider.""" 

2 

3from typing import Optional 

4import gws 

5import gws.lib.shape 

6import gws.lib.crs 

7import gws.lib.bounds 

8import gws.lib.jsonx 

9 

10 

11class Config(gws.Config): 

12 """Access to a GeoJSON file.""" 

13 

14 path: gws.FilePath 

15 """Path to a GeoJSON file.""" 

16 

17 

18class Object(gws.Node): 

19 """GeoJSON file provider. 

20 

21 Reads the features of a GeoJSON file into feature records and selects 

22 the records that match a search. 

23 """ 

24 

25 path: str 

26 """Path to the GeoJSON file.""" 

27 _records: list[gws.FeatureRecord] 

28 

29 def __getstate__(self): 

30 """Omit the loaded records when pickling.""" 

31 return gws.u.omit(vars(self), '_records') 

32 

33 def configure(self): 

34 self.path = self.cfg('path') 

35 

36 def cache_hash(self): 

37 """Return a hash that identifies the data of the provider. 

38 

39 Returns: 

40 A hash of the file path. 

41 """ 

42 return gws.u.sha256([self.path]) 

43 

44 def load_records(self): 

45 """Return all records of the file. 

46 

47 The file is read on the first call. 

48 

49 Returns: 

50 The feature records. 

51 """ 

52 if getattr(self, '_records', None) is None: 

53 self._records = self._load() 

54 return self._records 

55 

56 def get_records(self, search: gws.SearchQuery) -> list[gws.FeatureRecord]: 

57 """Return the records that match a search. 

58 

59 A record matches if its shape intersects the search shape, extended by 

60 the tolerance, or, without a shape, the search bounds; if one of its 

61 attribute values contains the keyword, ignoring case; and if its uid 

62 is one of the search uids. Criteria not set in the search are ignored. 

63 

64 Args: 

65 search: The search query. 

66 

67 Returns: 

68 The matching records. 

69 """ 

70 shape = None 

71 

72 if search.shape: 

73 shape = search.shape 

74 if search.tolerance: 

75 tol_value, tol_unit = search.tolerance 

76 if tol_unit == gws.Uom.px: 

77 tol_value *= search.resolution 

78 shape = shape.tolerance_polygon(tol_value) 

79 elif search.bounds: 

80 shape = gws.lib.shape.from_bounds(search.bounds) 

81 

82 return [rec for rec in self.load_records() if self._record_matches(rec, search, shape)] 

83 

84 def _record_matches(self, rec: gws.FeatureRecord, search: gws.SearchQuery, shape: Optional[gws.Shape]) -> bool: 

85 """Check if a record matches the search criteria.""" 

86 if shape: 

87 if not rec.shape or not rec.shape.intersects(shape): 

88 return False 

89 

90 if search.keyword: 

91 if all(search.keyword.lower() not in str(v).lower() for v in rec.attributes.values()): 

92 return False 

93 

94 if search.uids and rec.uid not in search.uids: 

95 return False 

96 

97 return True 

98 

99 def _load(self): 

100 """Read the GeoJSON file into feature records.""" 

101 js = gws.lib.jsonx.from_path(self.path) 

102 

103 crs = gws.lib.crs.WGS84 

104 if 'crs' in js: 

105 # https://geojson.org/geojson-spec#named-crs 

106 crs = gws.lib.crs.require(js['crs']['properties']['name']) 

107 

108 records = [] 

109 

110 for n, f in enumerate(js.get('features', []), 1): 

111 p = f.get('properties', {}) 

112 rec = gws.FeatureRecord(attributes=p) 

113 if f.get('geometry'): 

114 rec.shape = gws.lib.shape.from_geojson(f['geometry'], crs) 

115 rec.uid = p.get('id') or p.get('uid') or p.get('fid') or p.get('sid') or str(n) or '' 

116 records.append(rec) 

117 

118 return records