diff --git a/scripts/wk/clone/ddrescue.py b/scripts/wk/clone/ddrescue.py index 1c6b29ed..29610988 100644 --- a/scripts/wk/clone/ddrescue.py +++ b/scripts/wk/clone/ddrescue.py @@ -14,6 +14,8 @@ import shutil import subprocess import time +from typing import Any + import psutil import pytz @@ -338,16 +340,19 @@ class State(): """Object for tracking hardware diagnostic data.""" def __init__(self): self.block_pairs: list[BlockPair] = [] - self.destination = None - self.log_dir = None - self.mode = None - self.panes = {} - self.source = None - self.working_dir = None + self.destination: pathlib.Path | None = None + self.log_dir: pathlib.Path = log.format_log_path() + self.log_dir = self.log_dir.parent.joinpath( + f'ddrescue-TUI_{time.strftime("%Y-%m-%d_%H%M%S%z")}/', + ) + self.progress_out: pathlib.Path = self.log_dir.joinpath('progress.out') + self.mode: str = '?' + self.source: hw_disk.Disk | None = None + self.working_dir: pathlib.Path = pathlib.Path() self.ui: tui.TUI = tui.TUI('Source') self.ui.add_title_pane('Destination') - def _add_block_pair(self, source, destination) -> None: + def _add_block_pair(self, source: hw_disk.Disk, destination: pathlib.Path) -> None: """Add BlockPair object and run safety checks.""" self.block_pairs.append( BlockPair( @@ -363,7 +368,7 @@ class State(): description = self.source.path.name return pathlib.Path(f'{self.working_dir}/Clone_{description}.json') - def _load_settings(self, discard_unused_settings=False) -> dict: + def _load_settings(self, discard_unused_settings: bool = False) -> dict[Any, Any]: """Load settings from previous run, returns dict.""" settings = {} settings_file = self._get_clone_settings_path() @@ -412,7 +417,7 @@ class State(): # Done return settings - def _save_settings(self, settings) -> None: + def _save_settings(self, settings: dict[Any, Any]) -> None: """Save settings for future runs.""" settings_file = self._get_clone_settings_path() @@ -424,7 +429,7 @@ class State(): cli.print_error('Failed to save clone settings') raise std.GenericAbort() from err - def add_clone_block_pairs(self) -> None: + def add_clone_block_pairs(self) -> list[hw_disk.Disk]: """Add device to device block pairs and set settings if necessary.""" source_sep = get_partition_separator(self.source.path.name) dest_sep = get_partition_separator(self.destination.path.name) @@ -488,13 +493,16 @@ class State(): # Done return source_parts - def add_image_block_pairs(self, source_parts) -> None: + def add_image_block_pairs(self, source_parts: list[hw_disk.Disk]) -> None: """Add device to image file block pairs.""" for part in source_parts: - bp_dest = self.destination - self._add_block_pair(part, bp_dest) + self._add_block_pair(part, self.destination) - def confirm_selections(self, prompt_msg, source_parts=None) -> None: + def confirm_selections( + self, + prompt_msg: str, + source_parts: list[hw_disk.Disk] | None = None, + ) -> None: """Show selection details and prompt for confirmation.""" report = [] @@ -635,24 +643,18 @@ class State(): """Get total size of all block_pairs in bytes, returns int.""" return sum(pair.size for pair in self.block_pairs) - def init_recovery(self, docopt_args) -> None: + def init_recovery(self, docopt_args: dict[str, Any]) -> None: """Select source/dest and set env.""" cli.clear_screen() source_parts = [] # Set log - self.log_dir = log.format_log_path() - self.log_dir = pathlib.Path( - f'{self.log_dir.parent}/' - f'ddrescue-TUI_{time.strftime("%Y-%m-%d_%H%M%S%z")}/' - ) log.update_log_path( dest_dir=self.log_dir, dest_name='main', keep_history=True, timestamp=False, ) - self.progress_out = self.log_dir.joinpath('progress.out') self.ui.set_progress_file(self.progress_out) # Set mode @@ -755,18 +757,22 @@ class State(): settings['First Run'] = False self._save_settings(settings) - def pass_above_threshold(self, pass_name) -> bool: + def pass_above_threshold(self, pass_name: str) -> bool: """Check if all block_pairs meet the pass threshold, returns bool.""" threshold = cfg.ddrescue.AUTO_PASS_THRESHOLDS[pass_name] return all( p.get_percent_recovered() >= threshold for p in self.block_pairs ) - def pass_complete(self, pass_name) -> bool: + def pass_complete(self, pass_name: str) -> bool: """Check if all block_pairs completed pass_name, returns bool.""" return all(p.pass_complete(pass_name) for p in self.block_pairs) - def prep_destination(self, source_parts, dry_run=True) -> None: + def prep_destination( + self, + source_parts: list[hw_disk.Disk], + dry_run: bool = True, + ) -> None: """Prep destination as necessary.""" # TODO: Split into Linux and macOS # logical sector size is not easily found under macOS @@ -982,13 +988,13 @@ class State(): _f.write('\n'.join(debug.generate_object_report(_bp))) _f.write('\n') - def skip_pass(self, pass_name) -> None: + def skip_pass(self, pass_name: str) -> None: """Mark block_pairs as skipped if applicable.""" for pair in self.block_pairs: if pair.status[pass_name] == 'Pending': pair.status[pass_name] = 'Skipped' - def update_progress_pane(self, overall_status) -> None: + def update_progress_pane(self, overall_status: str) -> None: """Update progress pane.""" report = [] separator = '─────────────────────' @@ -2159,10 +2165,6 @@ def run_recovery(state: State, main_menu, settings_menu, dry_run=True) -> None: state.ui.remove_all_worker_panes() state.ui.clear_current_pane_height() - for pane in ('SMART', 'Journal'): - if pane in state.panes: - tmux.kill_pane(state.panes.pop(pane)) - # Show warning if nothing was done if not attempted_recovery: cli.print_warning('No actions performed') @@ -2220,7 +2222,7 @@ def select_disk(prompt_msg, skip_disk=None) -> hw_disk.Disk: return selected_disk -def select_disk_parts(prompt_msg, disk) -> hw_disk.Disk: +def select_disk_parts(prompt_msg, disk) -> list[hw_disk.Disk]: """Select disk parts from list, returns list of Disk().""" title = ansi.color_string('ddrescue TUI: Partition Selection', 'GREEN') title += f'\n\nDisk: {disk.path} {disk.description}' @@ -2336,7 +2338,7 @@ def select_path(prompt_msg) -> pathlib.Path: def set_mode(docopt_args) -> str: """Set mode from docopt_args or user selection, returns str.""" - mode = None + mode = '?' # Check docopt_args if docopt_args['clone']: