Coverage for paintmaskanimator/main_window_autosave.py: 82%

1"""Autosave / crash recovery; owned by ``MainWindow`` as ``window.autosave``. 

2 

3Owns the autosave timer and the crash-recovery flow, writing through the 

4window's ``canvas``, ``timeline`` and ``build_project_metadata``. 

5 

6A collaborator rather than a mixin -- see ``main_window_export.py`` for why. 

7""" 

8from typing import TYPE_CHECKING 

9from concurrent.futures import ThreadPoolExecutor 

10import copy 

11 

12from PySide6.QtCore import QTimer 

13from PySide6.QtWidgets import QMessageBox 

14 

15from . import config, preferences, project_io 

16from .i18n import tr 

17from .logging_setup import get_logger 

18 

19if TYPE_CHECKING: 

20 from .main_window import MainWindow 

21 

22log = get_logger(__name__) 

23 

24 

25class AutosaveController: 

26 """Owned by ``MainWindow`` as ``window.autosave``. 

27 

28 A collaborator rather than a mixin -- see main_window_export.py for why. 

29 """ 

30 

31 def __init__(self, window: "MainWindow"): 

32 self.window = window 

33 self._executor = None 

34 self._pending = None 

35 self._png_cache = project_io.PngCache() 

36 

37 """Periodic best-effort project snapshots + startup crash recovery.""" 

38 

39 def path(self): 

40 return config.config_dir() / "autosave.pmap" 

41 

42 def start(self): 

43 """Periodically snapshot the project so a crash doesn't lose work.""" 

44 self.window._autosave_timer = QTimer(self.window) 

45 self.window._autosave_timer.timeout.connect(self.request_save) 

46 self.apply_preferences() 

47 

48 def apply_preferences(self): 

49 """(Re)start or stop the timer from the autosave preferences.""" 

50 timer = self.window._autosave_timer 

51 timer.setInterval(preferences.autosave_interval_minutes() * 60 * 1000) 

52 if preferences.autosave_enabled(): 

53 timer.start() 

54 else: 

55 timer.stop() 

56 

57 def request_save(self): 

58 """Capture on the UI thread; compress and write outside the event loop.""" 

59 if getattr(self.window, "_closing", False): 

60 return 

61 if self._pending is not None and not self._pending.done(): 

62 return 

63 if self.window.canvas.drawing: 

64 # Do not capture a half-finished stroke or share an active painter. 

65 QTimer.singleShot(1000, self.request_save) 

66 return 

67 try: 

68 metadata = copy.deepcopy(self.window.build_project_metadata()) 

69 frames = [frame.clone() for frame in self.window.canvas.frames] 

70 sequence_archive = { 

71 key: layer.clone() 

72 for key, layer in self.window.canvas._sequence_archive.items() 

73 } 

74 path = self.path() 

75 if self._executor is None: 75 ↛ 81line 75 didn't jump to line 81 because the condition on line 75 was always true

76 self._executor = ThreadPoolExecutor( 

77 max_workers=1, thread_name_prefix="pma-autosave" 

78 ) 

79 # Cloned QImages share pixels until an edit detaches them. The worker 

80 # owns its PNG cache, so manual saves/opens cannot race with it. 

81 self._pending = self._executor.submit( 

82 self._write, path, metadata, frames, sequence_archive, self._png_cache 

83 ) 

84 except Exception: 

85 log.exception("autosave snapshot failed") 

86 

87 @staticmethod 

88 def _write(path, metadata, frames, sequence_archive, png_cache): 

89 try: 

90 project_io.write_project_archive( 

91 path, metadata, frames, sequence_archive, png_cache 

92 ) 

93 except Exception: 

94 log.exception("autosave failed") 

95 

96 def save(self): 

97 # Explicit synchronous saves must not collide with a periodic write. 

98 if self._pending is not None: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 self._pending.result() 

100 # Must never raise into the event loop — autosave is best-effort. 

101 try: 

102 project_io.write_project_archive( 

103 self.path(), 

104 self.window.build_project_metadata(), 

105 self.window.canvas.frames, 

106 self.window.canvas._sequence_archive, 

107 self.window.png_cache, 

108 ) 

109 except Exception: # noqa: BLE001 - best-effort; must never raise into the event loop 

110 # Autosave is best-effort and must never raise into the event loop, 

111 # so the broad catch is intentional; log the traceback instead of 

112 # printing it so it lands in the app log. 

113 log.exception("autosave failed") 

114 

115 def maybe_restore(self): 

116 """On startup, offer to restore a leftover autosave (likely a crash).""" 

117 path = self.path() 

118 try: 

119 if not path.exists() or path.stat().st_size == 0: 119 ↛ 123line 119 didn't jump to line 123 because the condition on line 119 was always true

120 return 

121 except OSError: 

122 return 

123 answer = QMessageBox.question( 

124 self.window, 

125 tr("作業の復元"), 

126 tr("前回のセッションが正常に終了しなかった可能性があります。\n" 

127 "自動保存された作業を復元しますか？"), 

128 ) 

129 if answer == QMessageBox.StandardButton.Yes: 

130 self.window.project.open(str(path)) 

131 else: 

132 self.clear() 

133 

134 def clear(self): 

135 # Finish an in-flight atomic replacement before deleting the snapshot; 

136 # otherwise it could reappear after a clean shutdown. 

137 if self._executor is not None: 

138 self._executor.shutdown(wait=True) 

139 self._executor = None 

140 self._pending = None 

141 self._png_cache.clear() 

142 try: 

143 self.path().unlink(missing_ok=True) 

144 except OSError: 

145 # A stale snapshot can trigger another recovery prompt on the next 

146 # launch, so preserve the failure details for diagnosis. 

147 log.warning("failed to remove autosave snapshot: %s", self.path(), exc_info=True)