Pylint cleanup for ddrescue-tui part 3

This commit is contained in:
2Shirt 2019-05-13 20:35:54 -06:00
parent 1f8b795e9a
commit 6a5bba5ed3
Signed by: 2Shirt
GPG key ID: 152FAC923B0E132C

View file

@ -554,6 +554,7 @@ class RecoveryState():
# Search for EToC delta
matches = re.findall(r'remaining time:.*$', text, re.MULTILINE)
if matches:
# pylint: disable=invalid-name
r = REGEX_REMAINING_TIME.search(matches[-1])
if r.group('na'):
self.etoc = 'N/A'
@ -705,10 +706,10 @@ def get_dir_report(dir_path):
return '\n'.join(output)
def get_size_in_bytes(s):
def get_size_in_bytes(size):
"""Convert size string from lsblk string to bytes, returns int."""
s = re.sub(r'(\d+\.?\d*)\s*([KMGTB])B?', r'\1 \2B', s, re.IGNORECASE)
return convert_to_bytes(s)
size = re.sub(r'(\d+\.?\d*)\s*([KMGTB])B?', r'\1 \2B', size, re.IGNORECASE)
return convert_to_bytes(size)
def get_formatted_status(label, data):
@ -733,19 +734,19 @@ def get_formatted_status(label, data):
return status
def get_status_color(s, t_success=99, t_warn=90):
def get_status_color(status, t_success=99, t_warn=90):
"""Get color based on status, returns str."""
color = COLORS['CLEAR']
p_recovered = -1
try:
p_recovered = float(s)
p_recovered = float(status)
except ValueError:
# Status is either in lists below or will default to red
pass
if s in ('Pending',) or str(s)[-2:] in (' b', 'Kb', 'Mb', 'Gb', 'Tb'):
if status == 'Pending' or str(status)[-2:] in (' b', 'Kb', 'Mb', 'Gb', 'Tb'):
color = COLORS['CLEAR']
elif s in ('Skipped', 'Unknown'):
elif status in ('Skipped', 'Unknown'):
color = COLORS['YELLOW']
elif p_recovered >= t_success:
color = COLORS['GREEN']
@ -825,6 +826,7 @@ def menu_ddrescue(source_path, dest_path, run_mode):
def menu_main(state):
# pylint: disable=too-many-branches,too-many-statements
"""Main menu is used to set ddrescue settings."""
checkmark = '*'
if 'DISPLAY' in global_vars['Env']:
@ -874,13 +876,13 @@ def menu_main(state):
elif selection == 'S':
# Set settings for pass
pass_settings = []
for k, v in state.settings.items():
if not v['Enabled']:
for option, option_data in state.settings.items():
if not option_data['Enabled']:
continue
if 'Value' in v:
pass_settings.append('{}={}'.format(k, v['Value']))
if 'Value' in option_data:
pass_settings.append('{}={}'.format(option, option_data['Value']))
else:
pass_settings.append(k)
pass_settings.append(option)
for opt in main_options:
if 'Auto' in opt['Base Name']:
auto_run = opt['Enabled']
@ -932,13 +934,15 @@ def menu_settings(state):
# Build menu
settings = []
for k, v in sorted(state.settings.items()):
if not v.get('Hidden', False):
settings.append({'Base Name': k, 'Flag': k})
for option, option_data in sorted(state.settings.items()):
if not option_data.get('Hidden', False):
settings.append({'Base Name': option, 'Flag': option})
actions = [{'Name': 'Main Menu', 'Letter': 'M'}]
# Show menu
while True:
# pylint: disable=invalid-name
# TODO: Clean up and/or replace with new menu-select function
for s in settings:
s['Name'] = '{}{}{}'.format(
s['Base Name'],
@ -984,16 +988,16 @@ def read_map_file(map_path):
# Parse output
for line in result.stdout.decode().splitlines():
m = re.match(
data = re.match(
r'^\s*(?P<key>\S+):.*\(\s*(?P<value>\d+\.?\d*)%.*', line.strip())
if m:
if data:
try:
map_data[m.group('key')] = float(m.group('value'))
map_data[data.group('key')] = float(data.group('value'))
except ValueError:
raise GenericError('Failed to read map data')
m = re.match(r'.*current status:\s+(?P<status>.*)', line.strip())
if m:
map_data['pass completed'] = bool(m.group('status') == 'finished')
data = re.match(r'.*current status:\s+(?P<status>.*)', line.strip())
if data:
map_data['pass completed'] = bool(data.group('status') == 'finished')
# Check if 100% done
try:
@ -1007,6 +1011,7 @@ def read_map_file(map_path):
def run_ddrescue(state, pass_settings):
# pylint: disable=too-many-branches,too-many-statements
"""Run ddrescue pass."""
return_code = -1
aborted = False
@ -1021,8 +1026,8 @@ def run_ddrescue(state, pass_settings):
# Create SMART monitor pane
state.smart_out = '{}/smart_{}.out'.format(
global_vars['TmpDir'], state.smart_source.name)
with open(state.smart_out, 'w') as f:
f.write('Initializing...')
with open(state.smart_out, 'w') as _f:
_f.write('Initializing...')
state.panes['SMART'] = tmux_split_window(
behind=True, lines=12, vertical=True, watch=state.smart_out)
@ -1066,11 +1071,11 @@ def run_ddrescue(state, pass_settings):
# Update SMART display (every 30 seconds)
if i % 30 == 0:
state.smart_source.get_smart_details()
with open(state.smart_out, 'w') as f:
with open(state.smart_out, 'w') as _f:
report = state.smart_source.generate_attribute_report(
timestamp=True)
for line in report:
f.write('{}\n'.format(line))
_f.write('{}\n'.format(line))
i += 1
# Update progress
@ -1135,6 +1140,8 @@ def run_ddrescue(state, pass_settings):
def select_parts(source_device):
# pylint: disable=too-many-branches
# TODO: Clean up and/or replace with new menu-select function
"""Select partition(s) or whole device, returns list of DevObj()s."""
selected_parts = []
children = source_device.details.get('children', [])
@ -1196,24 +1203,26 @@ def select_parts(source_device):
raise GenericAbort()
# Build list of selected parts
for d in dev_options:
if d['Selected']:
d['Dev'].model = source_device.model
d['Dev'].model_size = source_device.model_size
d['Dev'].update_filename_prefix()
selected_parts.append(d['Dev'])
for d_o in dev_options:
if d_o['Selected']:
d_o['Dev'].model = source_device.model
d_o['Dev'].model_size = source_device.model_size
d_o['Dev'].update_filename_prefix()
selected_parts.append(d_o['Dev'])
return selected_parts
def select_path(skip_device=None):
# pylint: disable=too-many-branches,too-many-locals
# TODO: Clean up and/or replace with new menu-select function
"""Optionally mount local dev and select path, returns DirObj."""
wd = os.path.realpath(global_vars['Env']['PWD'])
work_dir = os.path.realpath(global_vars['Env']['PWD'])
selected_path = None
# Build menu
path_options = [
{'Name': 'Current directory: {}'.format(wd), 'Path': wd},
{'Name': 'Current directory: {}'.format(work_dir), 'Path': work_dir},
{'Name': 'Local device', 'Path': None},
{'Name': 'Enter manually', 'Path': None}]
actions = [{'Name': 'Quit', 'Letter': 'Q'}]
@ -1228,9 +1237,9 @@ def select_path(skip_device=None):
raise GenericAbort()
elif selection.isnumeric():
index = int(selection) - 1
if path_options[index]['Path'] == wd:
if path_options[index]['Path'] == work_dir:
# Current directory
selected_path = DirObj(wd)
selected_path = DirObj(work_dir)
elif path_options[index]['Name'] == 'Local device':
# Local device
@ -1246,15 +1255,15 @@ def select_path(skip_device=None):
# Select volume
vol_options = []
for k, v in sorted(report.items()):
disabled = v['show_data']['data'] == 'Failed to mount'
for _k, _v in sorted(report.items()):
disabled = _v['show_data']['data'] == 'Failed to mount'
if disabled:
name = '{name} (Failed to mount)'.format(**v)
name = '{name} (Failed to mount)'.format(**_v)
else:
name = '{name} (mounted on "{mount_point}")'.format(**v)
name = '{name} (mounted on "{mount_point}")'.format(**_v)
vol_options.append({
'Name': name,
'Path': v['mount_point'],
'Path': _v['mount_point'],
'Disabled': disabled})
selection = menu_select(
title='Please select a volume',
@ -1329,10 +1338,11 @@ def select_device(description='device', skip_device=None):
action_entries=actions,
disabled_label='ALREADY SELECTED')
if selection == 'Q':
raise GenericAbort()
if selection.isnumeric():
return dev_options[int(selection)-1]['Dev']
elif selection == 'Q':
raise GenericAbort()
def setup_loopback_device(source_path):
@ -1371,6 +1381,7 @@ def show_selection_details(state):
def show_usage(script_name):
"""Show usage."""
print_info('Usage:')
print_standard(USAGE.format(script_name=script_name))
pause()
@ -1420,8 +1431,8 @@ def update_sidepane(state):
# Add line-endings
output = ['{}\n'.format(line) for line in output]
with open(state.progress_out, 'w') as f:
f.writelines(output)
with open(state.progress_out, 'w') as _f:
_f.writelines(output)
if __name__ == '__main__':