Coverage for paintmaskanimator/frame_scope.py: 91%
167 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-02 04:50 +0000
1"""フレーム横断の一括適用ランナー（Issue #9）。
3「すべてのコマに適用する」仕組みは、これまで機能ごとに書かれていた。ゴミ取りは
4``tools.dust_all_frames`` を、選択変形は ``tools.selection_all_frames`` と
5``canvas.transform_apply_all_frames`` を各自で参照し、進捗ダイアログの回し方も
6Undo のまとめ方もその場ごとに組み立てられていた。新しい操作を足すたびに
7「全コマ対応」を書き直す構造になっていた、ということ。
9このモジュールはその共通部分だけを引き受ける。呼び出し側が渡すのは
10**1レイヤーを受け取って結果画像を返す関数**（``op``）と、どのセルに適用するかを
11表す :class:`FrameScope` の2つだけ。走査順・保持コマの重複除去・共有画像の扱い・
12進捗通知・キャンセル・Undo へのまとめ上げは :func:`apply_over_scope` が持つ。
14UI から独立している（QWidget を触らない）ので、GUI なしで単体テストできる。
15``doc`` に要求するのは ``frames`` 属性だけで、``resolve_key_frame`` と
16``push_undo`` は在れば使う。
17"""
18from __future__ import annotations
20from dataclasses import dataclass, field
21from typing import Any, Callable, List, Optional, Tuple
23from .logging_setup import get_logger
24from .undo_entries import ScopedBatchUndo
26log = get_logger(__name__)
28__all__ = [
29 "FrameScope",
30 "LayerContext",
31 "ScopeResult",
32 "ScopeCancelled",
33 "apply_over_scope",
34 "resolve_cells",
35 "resolve_key_frame",
36]
39class ScopeCancelled(Exception):
40 """``op`` 側から実行を中断するための例外。部分適用は破棄される。"""
43def resolve_key_frame(frames, column, layer_index):
44 """``column`` のセルが参照している実体キーフレームの番号を返す。
46 ``canvas_key_frame.resolve_key_frame`` と同じ規則の、ドキュメントを必要と
47 しない純関数版。保持セルは実体を持つ手前のコマを指し、未使用セルと空キー
48 （``is_blank_key``）は ``None`` を返す。
49 """
50 if not frames: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 return None
52 column = max(0, min(int(column), len(frames) - 1))
53 for key_column in range(column, -1, -1): 53 ↛ 66line 53 didn't jump to line 66 because the loop on line 53 didn't complete
54 layers = frames[key_column].layers
55 if layer_index >= len(layers): 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 continue
57 layer = layers[layer_index]
58 if layer.has_content:
59 return (
60 key_column
61 if column < key_column + max(1, int(layer.exposure))
62 else None
63 )
64 if getattr(layer, "is_blank_key", False):
65 return None
66 return None
69@dataclass(frozen=True)
70class FrameScope:
71 """どのコマ・どのレイヤーに適用するか。
73 これまで各ツールが持っていた「すべてのコマ」チェックボックスの状態は、
74 最終的にこの1つの値へ集約される。
75 """
77 frames: Tuple[int, ...]
78 layers: Tuple[int, ...]
79 selection_only: bool = False
80 #: 保持セルを実体キーフレームへ畳むか。False にすると指定コマをそのまま辿る。
81 key_frames_only: bool = True
83 def __post_init__(self):
84 object.__setattr__(
85 self, "frames", tuple(int(value) for value in self.frames)
86 )
87 object.__setattr__(
88 self, "layers", tuple(int(value) for value in self.layers)
89 )
91 @classmethod
92 def current_frame(cls, doc, *, layers=None, selection_only=False):
93 """現在コマ×アクティブレイヤー（既定のスコープ）。"""
94 return cls(
95 (int(doc.current_frame),),
96 _default_layers(doc, layers),
97 selection_only,
98 )
100 @classmethod
101 def all_frames(cls, doc, *, layers=None, selection_only=False):
102 """全コマ×指定レイヤー（既定はアクティブレイヤー）。"""
103 return cls(
104 tuple(range(len(doc.frames))),
105 _default_layers(doc, layers),
106 selection_only,
107 )
109 @classmethod
110 def cells(cls, frames, layers, *, selection_only=False):
111 """タイムラインで選んだセル範囲などを、そのままスコープにする。"""
112 return cls(tuple(frames), tuple(layers), selection_only)
114 @classmethod
115 def choose(cls, doc, all_frames, *, layers=None, selection_only=False):
116 """真偽値1つで現在コマ／全コマを切り替える移行用のヘルパー。
118 既存の「すべてのコマ」チェックボックスを段階的に置き換えるための入口。
119 """
120 factory = cls.all_frames if all_frames else cls.current_frame
121 return factory(doc, layers=layers, selection_only=selection_only)
123 @property
124 def is_all_frames(self) -> bool:
125 return len(self.frames) > 1
128def _default_layers(doc, layers):
129 if layers is not None:
130 return tuple(int(value) for value in layers)
131 return (int(getattr(doc, "active_layer_index", 0)),)
134@dataclass
135class LayerContext:
136 """``op`` に渡る1セル分の入力。"""
138 doc: Any
139 frame: int
140 layer_index: int
141 layer: Any
142 selection_only: bool
143 #: このセルと画像実体を共有しているセル（``(frame, layer)``）。
144 shared_cells: Tuple[Tuple[int, int], ...] = ()
146 @property
147 def image(self):
148 return self.layer.image
151@dataclass
152class ScopeResult:
153 """1回の一括適用の結果。"""
155 label: str
156 changed_cells: int = 0
157 changed_pixels: int = 0
158 changed_frames: List[int] = field(default_factory=list)
159 #: 変更した ``(frame, layer)`` の一覧（レイヤーを跨ぐ通知のため）。
160 changed_cell_list: List[Tuple[int, int]] = field(default_factory=list)
161 undo: Optional[ScopedBatchUndo] = None
162 cancelled: bool = False
163 #: 走査したセル数（変更の有無を問わない）。
164 visited_cells: int = 0
166 @property
167 def changed(self) -> bool:
168 return self.changed_cells > 0
170 def __bool__(self) -> bool:
171 return self.changed
174def resolve_cells(doc, scope: FrameScope) -> List[Tuple[int, int]]:
175 """スコープを、実際に処理する ``(frame, layer)`` の一覧へ展開する。
177 保持セルは実体キーフレームへ畳み、同じセルは1度だけ返す。走査順はコマ番号
178 →レイヤー番号の昇順で安定させる（進捗表示とテストのため）。
179 """
180 frames = getattr(doc, "frames", None) or []
181 resolver = getattr(doc, "resolve_key_frame", None)
182 seen = set()
183 cells: List[Tuple[int, int]] = []
184 for layer_index in scope.layers:
185 if layer_index < 0: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 continue
187 for frame_index in scope.frames:
188 if not 0 <= frame_index < len(frames): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 continue
190 if scope.key_frames_only: 190 ↛ 200line 190 didn't jump to line 200 because the condition on line 190 was always true
191 if resolver is not None:
192 key_frame = resolver(int(frame_index), int(layer_index))
193 else:
194 key_frame = resolve_key_frame(
195 frames, frame_index, layer_index
196 )
197 if key_frame is None:
198 continue
199 else:
200 key_frame = int(frame_index)
201 if layer_index >= len(frames[key_frame].layers): 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 continue
203 cell = (int(key_frame), int(layer_index))
204 if cell in seen:
205 continue
206 seen.add(cell)
207 cells.append(cell)
208 cells.sort()
209 return cells
212def _group_shared_images(doc, cells):
213 """同じ絵（``Cell``）を指すセルをまとめる。
215 連番セルは ``coalesce_numbered_images`` で同じ ``Cell`` を共有し、
216 結果として同じ ``QImage`` を指す。セルごとに個別処理すると同じ絵を何度も
217 処理することになるので、共有グループは1回だけ処理して結果を全員へ配る。
219 判定は ``QImage`` の同一性で行う。``Cell`` の共有はこれを含むうえ、
220 取り込み経路などが ``Cell`` を通さず同じ画像を指す場合も拾えるため
221 （Issue #31 Phase 1 では、共有の判定をこれまでより狭めない）。
222 """
223 frames = doc.frames
224 groups: List[Tuple[Tuple[int, int], List[Tuple[int, int]]]] = []
225 index_of = {}
226 for frame_index, layer_index in cells:
227 layer = frames[frame_index].layers[layer_index]
228 key = id(layer.image)
229 position = index_of.get(key)
230 if position is None:
231 index_of[key] = len(groups)
232 groups.append(((frame_index, layer_index), [(frame_index, layer_index)]))
233 else:
234 groups[position][1].append((frame_index, layer_index))
235 return groups
238def apply_over_scope(
239 doc,
240 scope: FrameScope,
241 op: Callable[[LayerContext], Optional[Any]],
242 *,
243 label: str,
244 progress: Optional[Callable[[int, int, Tuple[int, int]], None]] = None,
245 should_cancel: Optional[Callable[[], bool]] = None,
246 push_undo: bool = True,
247 count_pixels: Optional[Callable[[LayerContext, Any], int]] = None,
248) -> ScopeResult:
249 """``scope`` の各セルへ ``op`` を適用する。
251 ``op`` は「1レイヤーを受け取り、変更後の画像を返す（変更なしなら ``None``）」
252 純粋な関数として書く。渡された ``layer.image`` を破壊的に書き換えてはいけない
253 ――キャンセル時に部分適用を残さないため、結果はすべての ``op`` が終わるまで
254 ドキュメントへ書き戻さない。
256 :param progress: ``(done, total, cell)`` を受け取る通知。UI 側が進捗ダイアログを
257 回すための唯一の接点。
258 :param should_cancel: 各セルの前に呼ばれ、``True`` なら中断する。中断時は
259 ドキュメントを一切変更せず ``cancelled=True`` の結果を返す。
260 :param count_pixels: 変更画素数をメッセージに出したい呼び出し元向け。省略時は
261 変更セル数だけを数える。
262 :returns: :class:`ScopeResult`。``push_undo`` が真なら Undo は
263 :class:`~.undo_entries.ScopedBatchUndo` 1件として積まれる（＝Undo 1回で全復元）。
264 """
265 result = ScopeResult(label=label)
266 cells = resolve_cells(doc, scope)
267 if not cells:
268 return result
270 groups = _group_shared_images(doc, cells)
271 total = len(groups)
272 pending: List[Tuple[Tuple[int, int], List[Tuple[int, int]], Any, int]] = []
274 for done, (lead_cell, shared_cells) in enumerate(groups):
275 if should_cancel is not None and should_cancel():
276 result.cancelled = True
277 return result
278 if progress is not None:
279 progress(done, total, lead_cell)
281 frame_index, layer_index = lead_cell
282 layer = doc.frames[frame_index].layers[layer_index]
283 context = LayerContext(
284 doc=doc,
285 frame=frame_index,
286 layer_index=layer_index,
287 layer=layer,
288 selection_only=scope.selection_only,
289 shared_cells=tuple(shared_cells),
290 )
291 try:
292 produced = op(context)
293 except ScopeCancelled:
294 result.cancelled = True
295 return result
297 result.visited_cells += len(shared_cells)
298 if produced is None:
299 continue
300 changed_pixels = (
301 int(count_pixels(context, produced)) if count_pixels is not None else 0
302 )
303 pending.append((lead_cell, shared_cells, produced, changed_pixels))
305 if progress is not None:
306 progress(total, total, cells[-1])
308 if not pending:
309 return result
311 undo_cells = []
312 for _lead_cell, shared_cells, produced, changed_pixels in pending:
313 for frame_index, layer_index in shared_cells:
314 layer = doc.frames[frame_index].layers[layer_index]
315 undo_cells.append((
316 int(frame_index),
317 int(layer_index),
318 layer.image.copy(),
319 bool(layer.has_content),
320 ))
321 # 同じ絵を指すセルは1つの Cell を共有しているので、書き込みは
322 # グループの先頭1回で全員に行き渡る。共有が切れていた場合に
323 # そなえ、ここでも結び直しておく。
324 layer.image = produced
325 layer.has_content = True
326 result.changed_cells += 1
327 result.changed_frames.append(int(frame_index))
328 result.changed_cell_list.append((int(frame_index), int(layer_index)))
329 result.changed_pixels += changed_pixels
331 if push_undo: 331 ↛ 338line 331 didn't jump to line 338 because the condition on line 331 was always true
332 entry = ScopedBatchUndo(undo_cells, label)
333 pusher = getattr(doc, "push_undo", None)
334 if pusher is not None: 334 ↛ 336line 334 didn't jump to line 336 because the condition on line 334 was always true
335 pusher(entry)
336 result.undo = entry
337 else:
338 result.undo = ScopedBatchUndo(undo_cells, label)
339 return result