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 pp = gws.lib.osx.parse_path(p) 

47 self.paths = sorted( 

48 gws.lib.osx.find_files( 

49 pp.dirname, 

50 fnmatch.translate(pp.filename), 

51 ) 

52 ) 

53 return 

54 

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

56 

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

58 es1 = [] 

59 

60 for path in self.paths: 

61 try: 

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

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

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

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

66 

67 if not es1: 

68 return [] 

69 

70 # all images must have the same CRS 

71 es2 = [] 

72 crs = es1[0].bounds.crs 

73 for e in es1: 

74 if e.bounds.crs == crs: 

75 es2.append(e) 

76 continue 

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

78 

79 return es2 

80 

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

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

83 

84 records = [] 

85 

86 for e in entries: 

87 records.append( 

88 gws.FeatureRecord( 

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

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

91 ) 

92 ) 

93 

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

95 la = ds.create_layer( 

96 name=file_name, 

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

98 geometry_type=gws.GeometryType.polygon, 

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

100 ) 

101 la.insert(records) 

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

103 

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

105 return idx_path