Coverage for gws-app/gws/plugin/raster_layer/provider.py: 84%

1"""Raster image provider.""" 

2 

3import fnmatch 

4from typing import Optional 

5 

6import gws 

7import gws.base.shape 

8import gws.lib.osx 

9import gws.lib.crs 

10import gws.gis.gdalx 

11 

12 

13class Config(gws.Config): 

14 """Raster data provider.""" 

15 

16 paths: Optional[list[gws.FilePath]] 

17 """List of image file paths.""" 

18 pathPattern: Optional[str] 

19 """Glob pattern for image file paths.""" 

20 crs: Optional[gws.CrsName] 

21 """Default CRS for the images.""" 

22 

23 

24class ImageEntry(gws.Data): 

25 path: str 

26 bounds: gws.Bounds 

27 

28 

29class Object(gws.Node): 

30 paths: list[str] 

31 crs: Optional[gws.Crs] 

32 

33 def configure(self): 

34 p = self.cfg('crs') 

35 self.crs = gws.lib.crs.require(p) if p else None 

36 

37 self.paths = [] 

38 

39 p = self.cfg('paths') 

40 if p: 

41 self.paths = p 

42 return 

43 

44 p = self.cfg('pathPattern') 

45 if p: 

46 v = gws.lib.osx.parse_path(p) 

47 self.paths = sorted(gws.lib.osx.find_files(v['dirname'], fnmatch.translate(v['filename']))) 

48 return 

49 

50 raise gws.ConfigurationError('no paths or pathPattern specified for raster provider.') 

51 

52 def enumerate_images(self, default_crs: gws.Crs) -> list[ImageEntry]: 

53 es1 = [] 

54 

55 for path in self.paths: 

56 try: 

57 with gws.gis.gdalx.open_raster(path, default_crs=default_crs) as gd: 

58 es1.append(ImageEntry(path=path, bounds=gd.bounds())) 

59 except gws.gis.gdalx.Error as exc: 

60 gws.log.warning(f'raster_provider: {path!r}: ERROR: ({exc})') 

61 

62 if not es1: 

63 return [] 

64 

65 # all images must have the same CRS 

66 es2 = [] 

67 crs = es1[0].bounds.crs 

68 for e in es1: 

69 if e.bounds.crs == crs: 

70 es2.append(e) 

71 continue 

72 gws.log.warning(f'raster_provider: {e.path!r}: ERROR: wrong crs {e.bounds.crs}, must be {crs}') 

73 

74 return es2 

75 

76 def make_tile_index(self, entries: list[ImageEntry], file_name: str) -> str: 

77 idx_path = f'{gws.c.OBJECT_CACHE_DIR}/{file_name}.shp' 

78 

79 records = [] 

80 

81 for e in entries: 

82 records.append( 

83 gws.FeatureRecord( 

84 attributes={'location': e.path}, 

85 shape=gws.base.shape.from_bounds(e.bounds), 

86 ) 

87 ) 

88 

89 with gws.gis.gdalx.open_vector(idx_path, 'w') as ds: 

90 la = ds.create_layer( 

91 name=file_name, 

92 columns={'location': gws.AttributeType.str}, 

93 geometry_type=gws.GeometryType.polygon, 

94 crs=entries[0].bounds.crs, 

95 ) 

96 la.insert(records) 

97 ds.gdDataset.ExecuteSQL(f'CREATE SPATIAL INDEX ON {file_name}') 

98 

99 gws.log.debug(f'raster_provider: created {idx_path=}') 

100 return idx_path