Coverage for gws-app/gws/gis/cache/cli.py: 0%

1"""Command-line cache commands.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.base.action 

7import gws.lib.uom 

8import gws.lib.cli as cli 

9import gws.config 

10 

11from . import core 

12 

13gws.ext.new.cli('cache') 

14 

15 

16class StatusParams(gws.CliParams): 

17 layer: Optional[list[str]] 

18 """list of layer IDs""" 

19 

20 

21class DropParams(gws.CliParams): 

22 layer: Optional[list[str]] 

23 """list of layer IDs""" 

24 

25 

26class SeedParams(gws.CliParams): 

27 layer: list[str] 

28 """list of layer IDs""" 

29 levels: list[int] 

30 """zoom levels to build the cache for""" 

31 

32 

33class Object(gws.Node): 

34 

35 @gws.ext.command.cli('cacheStatus') 

36 def do_status(self, p: StatusParams): 

37 """Display the cache status.""" 

38 

39 root = gws.config.loader.load() 

40 status = core.status(root, gws.u.to_list(p.layer)) 

41 

42 for e in status.entries: 

43 cli.info('') 

44 cli.info('=' * 80) 

45 

46 cli.info(f'CACHE {e.uid}') 

47 cli.info(f'DIR {e.dirname}') 

48 

49 ls = [] 

50 for la in e.layers: 

51 title = gws.u.get(la, 'title', '?') 

52 ls.append(f'{la.uid}: {title!r} type={la.extType}') 

53 cli.info(f'LAYER : {_comma(ls)}') 

54 

55 table = [] 

56 

57 for z, g in sorted(e.grids.items()): 

58 table.append({ 

59 'level': z, 

60 'scale': '1:' + str(round(gws.lib.uom.res_to_scale(g.res))), 

61 'grid': f'{g.maxX} x {g.maxY}', 

62 'total': g.totalTiles, 

63 'cached': g.cachedTiles, 

64 '%%': int(100 * (g.cachedTiles / g.totalTiles)), 

65 }) 

66 cli.info('') 

67 cli.info(cli.text_table(table, ['level', 'scale', 'grid', 'total', 'cached', '%%'])) 

68 

69 if status.staleDirs: 

70 cli.info('') 

71 cli.info('=' * 80) 

72 cli.info(f'{len(status.staleDirs)} STALE CACHES ("gws cache cleanup" to remove):') 

73 for d in status.staleDirs: 

74 cli.info(f' {d}') 

75 

76 @gws.ext.command.cli('cacheCleanup') 

77 def do_cleanup(self, p: gws.CliParams): 

78 """Remove stale cache dirs.""" 

79 

80 root = gws.config.loader.load() 

81 core.cleanup(root) 

82 

83 @gws.ext.command.cli('cacheDrop') 

84 def do_drop(self, p: DropParams): 

85 """Remove active cache dirs.""" 

86 

87 root = gws.config.loader.load() 

88 core.drop(root, gws.u.to_list(p.layer)) 

89 

90 @gws.ext.command.cli('cacheSeed') 

91 def do_seed(self, p: SeedParams): 

92 """Seed cache for layers.""" 

93 

94 root = gws.config.loader.load() 

95 status = core.status(root, gws.u.to_list(p.layer)) 

96 

97 levels = [] 

98 if p.levels: 

99 levels = [int(x) for x in gws.u.to_list(p.levels)] 

100 

101 core.seed(root, status.entries, levels) 

102 

103 

104_comma = ','.join 

105