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

1"""Image set provider for raster layers.""" 

2 

3import fnmatch 

4from typing import Optional 

5 

6import gws 

7import gws.lib.shape 

8import gws.lib.osx 

9import gws.lib.crs 

10import gws.lib.gdalx 

11 

12 

13class Config(gws.Config): 

14 """Set of georeferenced image files for a raster layer.""" 

15 

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

17 """Image files to show.""" 

18 pathPattern: Optional[str] 

19 """Glob pattern for image files.""" 

20 crs: Optional[gws.CrsName] 

21 """CRS for images that have none.""" 

22 

23 

24class ImageEntry(gws.Data): 

25 """A georeferenced image file.""" 

26 

27 path: str 

28 """File path.""" 

29 bounds: gws.Bounds 

30 """Image bounds in the image CRS.""" 

31 

32 

33class Object(gws.Node): 

34 """Set of georeferenced image files, given as paths or a glob pattern.""" 

35 

36 paths: list[str] 

37 """Image file paths.""" 

38 crs: Optional[gws.Crs] 

39 """CRS for images that have none.""" 

40 

41 def configure(self): 

42 p = self.cfg('crs') 

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

44 

45 self.paths = [] 

46 

47 p = self.cfg('paths') 

48 if p: 

49 self.paths = p 

50 return 

51 

52 p = self.cfg('pathPattern') 

53 if p: 

54 pp = gws.lib.osx.parse_path(p) 

55 self.paths = sorted( 

56 gws.lib.osx.find_files( 

57 pp.dirname, 

58 fnmatch.translate(pp.filename), 

59 ) 

60 ) 

61 return 

62 

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

64 

65 def cache_hash(self): 

66 """Compute a hash of the provider settings. 

67 

68 Returns: 

69 Hash string, built from the paths and the CRS. 

70 """ 

71 return gws.u.sha256([ 

72 self.paths, 

73 self.crs.srid if self.crs else '', 

74 ]) 

75 

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

77 """Read the bounds of the image files. 

78 

79 Files that cannot be opened, and files in a CRS other than the CRS of 

80 the first image, are skipped with a configuration warning. 

81 

82 Args: 

83 default_crs: CRS for images that have none. 

84 

85 Returns: 

86 Image entries, all in the same CRS. 

87 """ 

88 es1 = [] 

89 

90 for path in self.paths: 

91 try: 

92 with gws.lib.gdalx.open_raster(path, default_crs=default_crs) as gd: 

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

94 except gws.lib.gdalx.Error as exc: 

95 self.root.config_warning(f'raster_provider: {path!r}: cannot open: ({exc})') 

96 

97 if not es1: 

98 return [] 

99 

100 # all images must have the same CRS 

101 es2 = [] 

102 crs = es1[0].bounds.crs 

103 for e in es1: 

104 if e.bounds.crs == crs: 

105 es2.append(e) 

106 continue 

107 self.root.config_warning(f'raster_provider: {e.path!r}: wrong crs {e.bounds.crs}, must be {crs}') 

108 

109 return es2 

110 

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

112 """Create a MapServer tile index shapefile for the images. 

113 

114 The index has a polygon per image with the file path in the 

115 ``location`` column, and a spatial index. 

116 

117 Args: 

118 entries: Images to index. 

119 file_name: Base name of the shapefile, in the object cache directory. 

120 

121 Returns: 

122 Path to the shapefile. 

123 """ 

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

125 

126 records = [] 

127 

128 for e in entries: 

129 records.append( 

130 gws.FeatureRecord( 

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

132 shape=gws.lib.shape.from_bounds(e.bounds), 

133 ) 

134 ) 

135 

136 with gws.lib.gdalx.open_vector(idx_path, 'w') as ds: 

137 la = ds.create_layer( 

138 name=file_name, 

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

140 geometry_type=gws.GeometryType.polygon, 

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

142 ) 

143 la.insert(records) 

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

145 

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

147 return idx_path