Coverage for gws-app/gws/gis/cache/store.py: 97%

1"""Filesystem tile store, MapProxy 'mp' directory layout.""" 

2 

3import os 

4 

5import gws 

6import gws.lib.osx as osx 

7 

8 

9class Object(gws.TileStore): 

10 """Filesystem tile store.""" 

11 

12 def __init__(self, base_dir: str, max_age: int, extension: str): 

13 """Create a store. 

14 

15 Args: 

16 base_dir: Base directory. 

17 max_age: Max. age of stored tiles in seconds. 

18 extension: File extension of tile files, without a dot. 

19 """ 

20 self.baseDir = base_dir 

21 self.maxAge = max_age 

22 self.extension = extension 

23 

24 def stats(self) -> gws.TileStoreStats: 

25 return self._stats() 

26 

27 def stats_for_level(self, z: int) -> gws.TileStoreStats: 

28 return self._stats(z) 

29 

30 def path(self, mt: gws.MapTile) -> str: 

31 x, y, z = mt 

32 s = 10000 

33 return f'{self._level_dir(z)}/{(x // s):04d}/{(x % s):04d}/{(y // s):04d}/{(y % s):04d}.{self.extension}' 

34 

35 def has(self, mt: gws.MapTile) -> bool: 

36 p = self.path(mt) 

37 age = osx.file_age(p) 

38 return 0 <= age < self.maxAge 

39 

40 def read(self, mt: gws.MapTile) -> bytes | None: 

41 if not self.has(mt): 

42 return None 

43 try: 

44 with open(self.path(mt), 'rb') as fp: 

45 return fp.read() 

46 except OSError: 

47 return None 

48 

49 def write(self, mt: gws.MapTile, blob: bytes): 

50 p = self.path(mt) 

51 try: 

52 osx.mkdir(os.path.dirname(p)) 

53 gws.u.write_file_b(p, blob) 

54 except OSError as exc: 

55 gws.log.warning(f'tile store: write failed {p!r}: {exc}') 

56 

57 def drop(self): 

58 osx.rmdir(self.baseDir) 

59 

60 def drop_level(self, z: int): 

61 osx.rmdir(self._level_dir(z)) 

62 

63 def drop_range(self, mtr: gws.MapTileRange): 

64 x0, y0, x1, y1, z = mtr 

65 dir = self._level_dir(z) 

66 if not gws.u.is_dir(dir): 

67 return 

68 

69 for de in osx.find_entries(dir): 

70 if not de.is_file() or de.name.endswith('.tmp'): 

71 continue 

72 x, y = self._tile_xy(de.path) 

73 if x0 <= x <= x1 and y0 <= y <= y1: 

74 osx.unlink(de.path) 

75 

76 ## 

77 

78 def _level_dir(self, z: int) -> str: 

79 return f'{self.baseDir}/{z:02d}' 

80 

81 def _tile_xy(self, path: str) -> tuple[int, int]: 

82 a, b, c, d = path.rsplit('.', 1)[0].split('/')[-4:] 

83 return int(a) * 10000 + int(b), int(c) * 10000 + int(d) 

84 

85 def _stats(self, z: int | None = None) -> gws.TileStoreStats: 

86 stats = gws.TileStoreStats(count=0, size=0, range=None) 

87 dir = self._level_dir(z) if z is not None else self.baseDir 

88 if not gws.u.is_dir(dir): 

89 return stats 

90 

91 last_time = gws.u.stime() - self.maxAge 

92 

93 for de in osx.find_entries(dir): 

94 if not de.is_file() or de.name.endswith('.tmp'): 

95 continue 

96 s = de.stat() 

97 if s.st_size == 0 or s.st_mtime < last_time: 

98 continue 

99 stats.count += 1 

100 stats.size += s.st_size 

101 if z is None: 

102 continue 

103 

104 x, y = self._tile_xy(de.path) 

105 if stats.range is None: 

106 stats.range = (x, y, x, y, z) 

107 else: 

108 stats.range = ( 

109 min(stats.range[0], x), 

110 min(stats.range[1], y), 

111 max(stats.range[2], x), 

112 max(stats.range[3], y), 

113 z, 

114 ) 

115 

116 return stats