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

56 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-10-16 23:09 +0200

1"""GeoJSON provder.""" 

2 

3from typing import Optional 

4import gws 

5import gws.base.shape 

6import gws.lib.crs 

7import gws.lib.bounds 

8import gws.lib.jsonx 

9 

10 

11class Config(gws.Config): 

12 """Configuration for GeoJSON provider.""" 

13 

14 path: gws.FilePath 

15 """path to a GeoJSON file""" 

16 

17 

18class Object(gws.Node): 

19 path: str 

20 _records: list[gws.FeatureRecord] 

21 

22 def __getstate__(self): 

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

24 

25 def configure(self): 

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

27 

28 def load_records(self): 

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

30 self._records = self._load() 

31 return self._records 

32 

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

34 shape = None 

35 

36 if search.shape: 

37 shape = search.shape 

38 if search.tolerance: 

39 tol_value, tol_unit = search.tolerance 

40 if tol_unit == gws.Uom.px: 

41 tol_value *= search.resolution 

42 shape = shape.tolerance_polygon(tol_value) 

43 elif search.bounds: 

44 shape = gws.base.shape.from_bounds(search.bounds) 

45 

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

47 

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

49 if shape: 

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

51 return False 

52 

53 if search.keyword: 

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

55 return False 

56 

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

58 return False 

59 

60 return True 

61 

62 def _load(self): 

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

64 

65 crs = gws.lib.crs.WGS84 

66 if 'crs' in js: 

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

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

69 

70 records = [] 

71 

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

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

74 rec = gws.FeatureRecord(attributes=p) 

75 if f.get('geometry'): 

76 rec.shape = gws.base.shape.from_geojson(f['geometry'], crs) 

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

78 records.append(rec) 

79 

80 return records