990 lines
28 KiB
Python
990 lines
28 KiB
Python
"""WizardKit: Hardware diagnostics"""
|
|
# vim: sts=2 sw=2 ts=2
|
|
|
|
import atexit
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import time
|
|
|
|
from docopt import docopt
|
|
|
|
from wk import cfg, debug, exe, log, std, tmux
|
|
from wk.cfg.hw import STATUS_COLORS
|
|
from wk.hw import benchmark as hw_benchmark
|
|
from wk.hw import cpu as hw_cpu
|
|
from wk.hw import disk as hw_disk
|
|
from wk.hw import sensors as hw_sensors
|
|
from wk.hw import smart as hw_smart
|
|
from wk.hw import surface_scan as hw_surface_scan
|
|
from wk.hw import system as hw_system
|
|
from wk.hw.audio import audio_test
|
|
from wk.hw.keyboard import keyboard_test
|
|
from wk.hw.network import network_test
|
|
from wk.hw.screensavers import screensaver
|
|
from wk.hw.test import Test, TestGroup
|
|
|
|
|
|
# STATIC VARIABLES
|
|
DOCSTRING = f'''{cfg.main.KIT_NAME_FULL}: Hardware Diagnostics
|
|
|
|
Usage:
|
|
hw-diags [options]
|
|
hw-diags (-h | --help)
|
|
|
|
Options:
|
|
-c --cli Force CLI mode
|
|
-h --help Show this page
|
|
-q --quick Skip menu and perform a quick check
|
|
'''
|
|
LOG = logging.getLogger(__name__)
|
|
TEST_GROUPS = {
|
|
# Also used to build the menu options
|
|
## NOTE: This needs to be above MENU_SETS
|
|
'CPU & Cooling': 'cpu_stress_tests',
|
|
'Disk Attributes': 'disk_attribute_check',
|
|
'Disk Self-Test': 'disk_self_test',
|
|
'Disk Surface Scan': 'disk_surface_scan',
|
|
'Disk I/O Benchmark': 'disk_io_benchmark',
|
|
}
|
|
MENU_ACTIONS = (
|
|
'Audio Test',
|
|
'Keyboard Test',
|
|
'Network Test',
|
|
'Clock Sync',
|
|
'Start',
|
|
'Quit')
|
|
MENU_ACTIONS_SECRET = (
|
|
'Matrix',
|
|
'Tubes',
|
|
)
|
|
MENU_OPTIONS_QUICK = ('Disk Attributes',)
|
|
MENU_SETS = {
|
|
'Full Diagnostic': (*TEST_GROUPS,),
|
|
'Disk Diagnostic': (
|
|
'Disk Attributes',
|
|
'Disk Self-Test',
|
|
'Disk Surface Scan',
|
|
'Disk I/O Benchmark',
|
|
),
|
|
'Disk Diagnostic (Quick)': ('Disk Attributes',),
|
|
}
|
|
MENU_TOGGLES = (
|
|
'Skip USB Benchmarks',
|
|
)
|
|
PLATFORM = std.PLATFORM
|
|
|
|
# Classes
|
|
class State():
|
|
"""Object for tracking hardware diagnostic data."""
|
|
def __init__(self):
|
|
self.disks = []
|
|
self.layout = cfg.hw.TMUX_LAYOUT.copy()
|
|
self.log_dir = None
|
|
self.panes = {}
|
|
self.system = None
|
|
self.test_groups = []
|
|
self.top_text = std.color_string('Hardware Diagnostics', 'GREEN')
|
|
|
|
# Init tmux and start a background process to maintain layout
|
|
self.init_tmux()
|
|
exe.start_thread(self.fix_tmux_layout_loop)
|
|
|
|
def abort_testing(self) -> None:
|
|
"""Set unfinished tests as aborted and cleanup tmux panes."""
|
|
for group in self.test_groups:
|
|
for test in group.test_objects:
|
|
if test.status in ('Pending', 'Working'):
|
|
test.set_status('Aborted')
|
|
|
|
# Cleanup tmux
|
|
self.panes.pop('Current', None)
|
|
for key, pane_ids in self.panes.copy().items():
|
|
if key in ('Top', 'Started', 'Progress'):
|
|
continue
|
|
if isinstance(pane_ids, str):
|
|
tmux.kill_pane(self.panes.pop(key))
|
|
else:
|
|
for _id in pane_ids:
|
|
tmux.kill_pane(_id)
|
|
self.panes.pop(key)
|
|
|
|
def disk_safety_checks(self, prep=False, wait_for_self_tests=True) -> None:
|
|
# pylint: disable=too-many-branches,too-many-statements
|
|
"""Run disk safety checks."""
|
|
self_tests_in_progress = False
|
|
for disk in self.disks:
|
|
disable_tests = False
|
|
|
|
# Skip already disabled devices
|
|
if all(test.disabled for test in disk.tests):
|
|
continue
|
|
|
|
try:
|
|
hw_smart.safety_checks(disk)
|
|
except hw_smart.CriticalHardwareError:
|
|
disable_tests = True
|
|
disk.add_note('Critical hardware error detected.', 'RED')
|
|
if 'Disk Attributes' in disk.tests:
|
|
disk.tests['Disk Attributes'].failed = True
|
|
disk.tests['Disk Attributes'].set_status('Failed')
|
|
if not prep:
|
|
# Mid-diag failure detected
|
|
LOG.warning('Critical hardware error detected during diagnostics')
|
|
disk.add_note(
|
|
'Critical hardware error detected during diagnostics',
|
|
'YELLOW',
|
|
)
|
|
except hw_smart.SMARTSelfTestInProgressError as err:
|
|
if prep:
|
|
std.print_warning(f'SMART self-test(s) in progress for {disk.path}')
|
|
if std.ask('Continue with all tests disabled for this device?'):
|
|
disable_tests = True
|
|
else:
|
|
std.print_standard('Diagnostics aborted.')
|
|
std.print_standard(' ')
|
|
std.pause('Press Enter to exit...')
|
|
raise SystemExit(1) from err
|
|
elif wait_for_self_tests:
|
|
self_tests_in_progress = True
|
|
else:
|
|
# Other tests will NOT be disabled
|
|
LOG.warning('SMART data may not be reliable for: %s', disk.path)
|
|
# Add note to report
|
|
if 'Disk Self-Test' in disk.tests:
|
|
disk.tests['Disk Self-Test'].failed = True
|
|
disk.tests['Disk Self-Test'].report.append(
|
|
std.color_string('Please manually review SMART data', 'YELLOW'),
|
|
)
|
|
else:
|
|
if (
|
|
'Disk Attributes' in disk.tests
|
|
and not disk.tests['Disk Attributes'].failed
|
|
and not hw_smart.check_attributes(disk, only_blocking=False)
|
|
):
|
|
# No blocking errors encountered, but found minor attribute failures
|
|
if not prep:
|
|
# Mid-diag failure detected
|
|
LOG.warning('Attribute(s) failure detected during diagnostics')
|
|
disk.add_note(
|
|
'Attribute(s) failure detected during diagnostics',
|
|
'YELLOW',
|
|
)
|
|
disk.tests['Disk Attributes'].failed = True
|
|
disk.tests['Disk Attributes'].set_status('Failed')
|
|
|
|
# Check Surface Scan
|
|
if (
|
|
'Disk Surface Scan' in disk.tests
|
|
and disk.tests['Disk Surface Scan'].failed
|
|
and 'Disk I/O Benchmark' in disk.tests
|
|
):
|
|
# Disable I/O Benchmark test
|
|
disk.tests['Disk I/O Benchmark'].set_status('Skipped')
|
|
disk.tests['Disk I/O Benchmark'].disabled = True
|
|
|
|
# Disable tests if necessary
|
|
if disable_tests:
|
|
disk.disable_disk_tests()
|
|
|
|
# Wait for self-test(s)
|
|
if self_tests_in_progress:
|
|
std.print_warning('SMART self-test(s) in progress')
|
|
std.print_standard('Waiting 60 seconds before continuing...')
|
|
std.sleep(60)
|
|
self.disk_safety_checks(wait_for_self_tests=False)
|
|
|
|
def fix_tmux_layout(self, forced=True) -> None:
|
|
"""Fix tmux layout based on cfg.hw.TMUX_LAYOUT."""
|
|
try:
|
|
tmux.fix_layout(self.panes, self.layout, forced=forced)
|
|
except RuntimeError:
|
|
# Assuming self.panes changed while running
|
|
pass
|
|
|
|
def fix_tmux_layout_loop(self) -> None:
|
|
"""Fix tmux layout on a loop.
|
|
|
|
NOTE: This should be called as a thread.
|
|
"""
|
|
while True:
|
|
self.fix_tmux_layout(forced=False)
|
|
std.sleep(1)
|
|
|
|
def init_diags(self, menu) -> None:
|
|
"""Initialize diagnostic pass."""
|
|
|
|
# Reset objects
|
|
self.disks.clear()
|
|
self.layout.clear()
|
|
self.layout.update(cfg.hw.TMUX_LAYOUT)
|
|
self.test_groups.clear()
|
|
|
|
# Set log
|
|
self.log_dir = log.format_log_path()
|
|
self.log_dir = pathlib.Path(
|
|
f'{self.log_dir.parent}/'
|
|
f'Hardware-Diagnostics_{time.strftime("%Y-%m-%d_%H%M%S%z")}/'
|
|
)
|
|
log.update_log_path(
|
|
dest_dir=self.log_dir,
|
|
dest_name='main',
|
|
keep_history=False,
|
|
timestamp=False,
|
|
)
|
|
std.clear_screen()
|
|
std.print_info('Initializing...')
|
|
|
|
# Progress Pane
|
|
self.update_progress_pane()
|
|
tmux.respawn_pane(
|
|
pane_id=self.panes['Progress'],
|
|
watch_file=f'{self.log_dir}/progress.out',
|
|
)
|
|
|
|
# Add HW Objects
|
|
self.system = hw_system.System()
|
|
self.disks = hw_disk.get_disks(skip_kits=True)
|
|
|
|
# Add test objects
|
|
for name, details in menu.options.items():
|
|
if not details['Selected']:
|
|
# Only add selected options
|
|
continue
|
|
|
|
if 'CPU' in name:
|
|
# Create two Test objects which will both be used by cpu_stress_tests
|
|
# NOTE: Prime95 should be added first
|
|
self.system.tests.append(
|
|
Test(dev=self.system, label='Prime95', name=name),
|
|
)
|
|
self.system.tests.append(
|
|
Test(dev=self.system, label='Cooling', name=name),
|
|
)
|
|
self.test_groups.append(
|
|
TestGroup(
|
|
name=name,
|
|
function=globals()[TEST_GROUPS[name]],
|
|
test_objects=self.system.tests,
|
|
),
|
|
)
|
|
|
|
if 'Disk' in name:
|
|
test_group = TestGroup(
|
|
name=name, function=globals()[TEST_GROUPS[name]],
|
|
)
|
|
for disk in self.disks:
|
|
test_obj = Test(dev=disk, label=disk.path.name, name=name)
|
|
disk.tests.append(test_obj)
|
|
test_group.test_objects.append(test_obj)
|
|
self.test_groups.append(test_group)
|
|
|
|
# Run safety checks
|
|
self.disk_safety_checks(prep=True)
|
|
|
|
def init_tmux(self) -> None:
|
|
"""Initialize tmux layout."""
|
|
tmux.kill_all_panes()
|
|
|
|
# Top
|
|
self.panes['Top'] = tmux.split_window(
|
|
behind=True,
|
|
lines=2,
|
|
vertical=True,
|
|
text=f'{self.top_text}\nMain Menu',
|
|
)
|
|
|
|
# Started
|
|
self.panes['Started'] = tmux.split_window(
|
|
lines=cfg.hw.TMUX_SIDE_WIDTH,
|
|
target_id=self.panes['Top'],
|
|
text=std.color_string(
|
|
['Started', time.strftime("%Y-%m-%d %H:%M %Z")],
|
|
['BLUE', None],
|
|
sep='\n',
|
|
),
|
|
)
|
|
|
|
# Progress
|
|
self.panes['Progress'] = tmux.split_window(
|
|
lines=cfg.hw.TMUX_SIDE_WIDTH,
|
|
text=' ',
|
|
)
|
|
|
|
def save_debug_reports(self) -> None:
|
|
"""Save debug reports to disk."""
|
|
LOG.info('Saving debug reports')
|
|
debug_dir = pathlib.Path(f'{self.log_dir}/debug')
|
|
if not debug_dir.exists():
|
|
debug_dir.mkdir()
|
|
|
|
# State (self)
|
|
std.save_pickles({'state': self}, debug_dir)
|
|
with open(f'{debug_dir}/state.report', 'a', encoding='utf-8') as _f:
|
|
_f.write('\n'.join(debug.generate_object_report(self)))
|
|
|
|
# Disks
|
|
for disk in self.disks:
|
|
with open(
|
|
f'{debug_dir}/disk_{disk.path.name}.report', 'a',
|
|
encoding='utf-8') as _f:
|
|
_f.write('\n'.join(debug.generate_object_report(disk)))
|
|
_f.write('\n\n[Tests]')
|
|
for test in disk.tests:
|
|
_f.write(f'\n{test.name}:\n')
|
|
_f.write('\n'.join(debug.generate_object_report(test, indent=1)))
|
|
|
|
# SMC
|
|
if os.path.exists('/.wk-live-macos'):
|
|
data = []
|
|
try:
|
|
proc = exe.run_program(['smc', '-f'])
|
|
data.extend(proc.stdout.splitlines())
|
|
data.append('----')
|
|
proc = exe.run_program(['smc', '-l'])
|
|
data.extend(proc.stdout.splitlines())
|
|
except Exception: # pylint: disable=broad-except
|
|
LOG.ERROR('Error(s) encountered while exporting SMC data')
|
|
data = [line.strip() for line in data]
|
|
with open(f'{debug_dir}/smc.data', 'a', encoding='utf-8') as _f:
|
|
_f.write('\n'.join(data))
|
|
|
|
# System
|
|
with open(f'{debug_dir}/system.report', 'a', encoding='utf-8') as _f:
|
|
_f.write('\n'.join(debug.generate_object_report(self.system)))
|
|
_f.write('\n\n[Tests]')
|
|
for test in self.system.tests:
|
|
_f.write(f'\n{test.name}:\n')
|
|
_f.write('\n'.join(debug.generate_object_report(test, indent=1)))
|
|
|
|
def update_clock(self) -> None:
|
|
"""Update 'Started' pane following clock sync."""
|
|
tmux.respawn_pane(
|
|
pane_id=self.panes['Started'],
|
|
text=std.color_string(
|
|
['Started', time.strftime("%Y-%m-%d %H:%M %Z")],
|
|
['BLUE', None],
|
|
sep='\n',
|
|
),
|
|
)
|
|
|
|
def update_progress_pane(self) -> None:
|
|
"""Update progress pane."""
|
|
report = []
|
|
width = cfg.hw.TMUX_SIDE_WIDTH
|
|
|
|
for group in self.test_groups:
|
|
report.append(std.color_string(group.name, 'BLUE'))
|
|
for test in group.test_objects:
|
|
report.append(std.color_string(
|
|
[test.label, f'{test.status:>{width-len(test.label)}}'],
|
|
[None, STATUS_COLORS.get(test.status, None)],
|
|
sep='',
|
|
))
|
|
|
|
# Add spacer
|
|
report.append(' ')
|
|
|
|
# Write to progress file
|
|
out_path = pathlib.Path(f'{self.log_dir}/progress.out')
|
|
with open(out_path, 'w', encoding='utf-8') as _f:
|
|
_f.write('\n'.join(report))
|
|
|
|
def update_top_pane(self, text) -> None:
|
|
"""Update top pane with text."""
|
|
tmux.respawn_pane(self.panes['Top'], text=f'{self.top_text}\n{text}')
|
|
|
|
|
|
# Functions
|
|
def build_menu(cli_mode=False, quick_mode=False) -> std.Menu:
|
|
# pylint: disable=too-many-branches
|
|
"""Build main menu, returns wk.std.Menu."""
|
|
menu = std.Menu(title=None)
|
|
|
|
# Add actions, options, etc
|
|
for action in MENU_ACTIONS:
|
|
menu.add_action(action)
|
|
for action in MENU_ACTIONS_SECRET:
|
|
menu.add_action(action, {'Hidden': True})
|
|
for option in TEST_GROUPS:
|
|
menu.add_option(option, {'Selected': True})
|
|
for toggle in MENU_TOGGLES:
|
|
menu.add_toggle(toggle, {'Selected': True})
|
|
for name, targets in MENU_SETS.items():
|
|
menu.add_set(name, {'Targets': targets})
|
|
menu.actions['Start']['Separator'] = True
|
|
|
|
# Update default selections for quick mode if necessary
|
|
if quick_mode:
|
|
for name in menu.options:
|
|
# Only select quick option(s)
|
|
menu.options[name]['Selected'] = name in MENU_OPTIONS_QUICK
|
|
|
|
# Skip CPU tests for TestStations
|
|
if os.path.exists(cfg.hw.TESTSTATION_FILE):
|
|
menu.options['CPU & Cooling']['Selected'] = False
|
|
|
|
# Add CLI actions if necessary
|
|
if cli_mode or 'DISPLAY' not in os.environ:
|
|
menu.add_action('Reboot')
|
|
menu.add_action('Power Off')
|
|
|
|
# Compatibility checks
|
|
if PLATFORM != 'Linux':
|
|
for name in ('Audio Test', 'Keyboard Test'):
|
|
menu.actions[name]['Disabled'] = True
|
|
if PLATFORM not in ('Darwin', 'Linux'):
|
|
for name in ('Matrix', 'Network Test', 'Tubes'):
|
|
menu.actions[name]['Disabled'] = True
|
|
|
|
# Live macOS actions
|
|
if os.path.exists('/.wk-live-macos'):
|
|
menu.actions['Clock Sync']['Separator'] = True
|
|
else:
|
|
menu.actions['Clock Sync']['Disabled'] = True
|
|
menu.actions['Clock Sync']['Hidden'] = True
|
|
|
|
# Done
|
|
return menu
|
|
|
|
|
|
def cpu_stress_tests(state, test_objects) -> None:
|
|
# pylint: disable=too-many-statements
|
|
"""CPU & cooling check using Prime95 and Sysbench."""
|
|
LOG.info('CPU Test (Prime95)')
|
|
aborted = False
|
|
prime_log = pathlib.Path(f'{state.log_dir}/prime.log')
|
|
run_sysbench = False
|
|
sensors_out = pathlib.Path(f'{state.log_dir}/sensors.out')
|
|
test_mprime_obj, test_cooling_obj = test_objects
|
|
|
|
# Bail early
|
|
if test_cooling_obj.disabled or test_mprime_obj.disabled:
|
|
return
|
|
|
|
# Prep
|
|
state.update_top_pane(test_mprime_obj.dev.cpu_description)
|
|
test_cooling_obj.set_status('Working')
|
|
test_mprime_obj.set_status('Working')
|
|
|
|
# Start sensors monitor
|
|
sensors = hw_sensors.Sensors()
|
|
sensors.start_background_monitor(
|
|
sensors_out,
|
|
thermal_action=('killall', 'mprime', '-INT'),
|
|
)
|
|
|
|
# Create monitor and worker panes
|
|
state.update_progress_pane()
|
|
state.panes['Prime95'] = tmux.split_window(
|
|
lines=10, vertical=True, watch_file=prime_log, watch_cmd='tail')
|
|
if PLATFORM == 'Darwin':
|
|
state.panes['Temps'] = tmux.split_window(
|
|
behind=True, percent=80, vertical=True, cmd='./hw-sensors')
|
|
elif PLATFORM == 'Linux':
|
|
state.panes['Temps'] = tmux.split_window(
|
|
behind=True, percent=80, vertical=True, watch_file=sensors_out)
|
|
tmux.resize_pane(height=3)
|
|
state.panes['Current'] = ''
|
|
state.layout['Current'] = {'height': 3, 'Check': True}
|
|
|
|
# Get idle temps
|
|
std.print_standard('Saving idle temps...')
|
|
sensors.save_average_temps(temp_label='Idle', seconds=5)
|
|
|
|
# Stress CPU
|
|
std.print_info('Running stress test')
|
|
hw_cpu.set_apple_fan_speed('max')
|
|
proc_mprime = hw_cpu.start_mprime(state.log_dir, prime_log)
|
|
|
|
# Show countdown
|
|
print('')
|
|
try:
|
|
print_countdown(proc=proc_mprime, seconds=cfg.hw.CPU_TEST_MINUTES*60)
|
|
except KeyboardInterrupt:
|
|
aborted = True
|
|
|
|
# Stop Prime95
|
|
hw_cpu.stop_mprime(proc_mprime)
|
|
|
|
# Update progress if necessary
|
|
if sensors.cpu_reached_critical_temp() or aborted:
|
|
test_cooling_obj.set_status('Aborted')
|
|
test_mprime_obj.set_status('Aborted')
|
|
state.update_progress_pane()
|
|
|
|
# Get cooldown temp
|
|
std.clear_screen()
|
|
std.print_standard('Letting CPU cooldown...')
|
|
std.sleep(5)
|
|
std.print_standard('Saving cooldown temps...')
|
|
sensors.save_average_temps(temp_label='Cooldown', seconds=5)
|
|
|
|
# Check Prime95 results
|
|
test_mprime_obj.report.append(std.color_string('Prime95', 'BLUE'))
|
|
hw_cpu.check_mprime_results(
|
|
test_obj=test_mprime_obj, working_dir=state.log_dir,
|
|
)
|
|
|
|
# Run Sysbench test if necessary
|
|
run_sysbench = (
|
|
not aborted and sensors.cpu_max_temp() >= cfg.hw.CPU_FAILURE_TEMP
|
|
)
|
|
if run_sysbench:
|
|
LOG.info('CPU Test (Sysbench)')
|
|
std.print_standard('Letting CPU cooldown more...')
|
|
std.sleep(30)
|
|
std.clear_screen()
|
|
std.print_info('Running alternate stress test')
|
|
print('')
|
|
proc_sysbench, filehandle_sysbench = hw_cpu.start_sysbench(
|
|
sensors,
|
|
sensors_out,
|
|
log_path=prime_log.with_name('sysbench.log'),
|
|
pane=state.panes['Prime95'],
|
|
)
|
|
try:
|
|
print_countdown(proc=proc_sysbench, seconds=cfg.hw.CPU_TEST_MINUTES*60)
|
|
except AttributeError:
|
|
# Assuming the sysbench process wasn't found and proc was set to None
|
|
LOG.error('Failed to find sysbench process', exc_info=True)
|
|
except KeyboardInterrupt:
|
|
aborted = True
|
|
hw_cpu.stop_sysbench(proc_sysbench, filehandle_sysbench)
|
|
|
|
# Update progress
|
|
# NOTE: CPU critical temp check isn't really necessary
|
|
# Hard to imagine it wasn't hit during Prime95 but was in sysbench
|
|
if sensors.cpu_reached_critical_temp() or aborted:
|
|
test_cooling_obj.set_status('Aborted')
|
|
test_mprime_obj.set_status('Aborted')
|
|
state.update_progress_pane()
|
|
|
|
# Check Cooling results
|
|
test_cooling_obj.report.append(std.color_string('Temps', 'BLUE'))
|
|
hw_cpu.check_cooling_results(test_cooling_obj, sensors, run_sysbench)
|
|
|
|
# Cleanup
|
|
state.update_progress_pane()
|
|
sensors.stop_background_monitor()
|
|
state.panes.pop('Current', None)
|
|
tmux.kill_pane(state.panes.pop('Prime95', None))
|
|
tmux.kill_pane(state.panes.pop('Temps', None))
|
|
|
|
# Done
|
|
if aborted:
|
|
raise std.GenericAbort('Aborted')
|
|
|
|
|
|
def disk_attribute_check(state, test_objects) -> None:
|
|
"""Disk attribute check."""
|
|
LOG.info('Disk Attribute Check')
|
|
for test in test_objects:
|
|
if not test.dev.attributes:
|
|
# No NVMe/SMART data
|
|
test.set_status('N/A')
|
|
continue
|
|
|
|
if hw_smart.check_attributes(test.dev):
|
|
test.passed = True
|
|
test.set_status('Passed')
|
|
else:
|
|
test.failed = True
|
|
test.set_status('Failed')
|
|
|
|
# Done
|
|
state.update_progress_pane()
|
|
|
|
|
|
def disk_io_benchmark(state, test_objects, skip_usb=True) -> None:
|
|
"""Disk I/O benchmark using dd."""
|
|
LOG.info('Disk I/O Benchmark (dd)')
|
|
aborted = False
|
|
|
|
# Run benchmarks
|
|
state.update_top_pane(
|
|
f'Disk I/O Benchmark{"s" if len(test_objects) > 1 else ""}',
|
|
)
|
|
state.panes['I/O Benchmark'] = tmux.split_window(
|
|
percent=50,
|
|
vertical=True,
|
|
text=' ',
|
|
)
|
|
for test in test_objects:
|
|
if test.disabled:
|
|
# Skip
|
|
continue
|
|
|
|
# Skip USB devices if requested
|
|
if skip_usb and test.dev.bus == 'USB':
|
|
test.set_status('Skipped')
|
|
continue
|
|
|
|
# Start benchmark
|
|
std.clear_screen()
|
|
std.print_report(test.dev.generate_report())
|
|
test.set_status('Working')
|
|
test_log = f'{state.log_dir}/{test.dev.path.name}_benchmark.out'
|
|
tmux.respawn_pane(
|
|
state.panes['I/O Benchmark'],
|
|
watch_cmd='tail',
|
|
watch_file=test_log,
|
|
)
|
|
state.update_progress_pane()
|
|
try:
|
|
hw_benchmark.run_io_test(test, test_log)
|
|
except KeyboardInterrupt:
|
|
aborted = True
|
|
except (subprocess.CalledProcessError, TypeError, ValueError) as err:
|
|
# Something went wrong
|
|
LOG.error('%s', err)
|
|
test.set_status('ERROR')
|
|
test.report.append(std.color_string(' Unknown Error', 'RED'))
|
|
|
|
# Mark test(s) aborted if necessary
|
|
if aborted:
|
|
test.set_status('Aborted')
|
|
test.report.append(std.color_string(' Aborted', 'YELLOW'))
|
|
break
|
|
|
|
# Update progress after each test
|
|
state.update_progress_pane()
|
|
|
|
# Cleanup
|
|
state.update_progress_pane()
|
|
tmux.kill_pane(state.panes.pop('I/O Benchmark', None))
|
|
|
|
# Done
|
|
if aborted:
|
|
raise std.GenericAbort('Aborted')
|
|
|
|
|
|
def disk_self_test(state, test_objects) -> None:
|
|
"""Disk self-test if available."""
|
|
LOG.info('Disk Self-Test(s)')
|
|
aborted = False
|
|
threads = []
|
|
state.panes['SMART'] = []
|
|
|
|
# Run self-tests
|
|
state.update_top_pane(
|
|
f'Disk self-test{"s" if len(test_objects) > 1 else ""}',
|
|
)
|
|
std.print_info(f'Starting self-test{"s" if len(test_objects) > 1 else ""}')
|
|
for test in reversed(test_objects):
|
|
if test.disabled:
|
|
# Skip
|
|
continue
|
|
|
|
# Start thread
|
|
test.set_status('Working')
|
|
test_log = f'{state.log_dir}/{test.dev.path.name}_selftest.log'
|
|
threads.append(exe.start_thread(hw_smart.run_self_test, args=(test, test_log)))
|
|
|
|
# Show progress
|
|
if threads[-1].is_alive():
|
|
state.panes['SMART'].append(
|
|
tmux.split_window(lines=4, vertical=True, watch_file=test_log),
|
|
)
|
|
|
|
# Wait for all tests to complete
|
|
state.update_progress_pane()
|
|
try:
|
|
while True:
|
|
if any(t.is_alive() for t in threads):
|
|
std.sleep(1)
|
|
else:
|
|
break
|
|
except KeyboardInterrupt:
|
|
aborted = True
|
|
for test in test_objects:
|
|
hw_smart.abort_self_test(test.dev)
|
|
std.sleep(0.5)
|
|
|
|
# Save report(s)
|
|
for test in test_objects:
|
|
hw_smart.check_self_test_results(test, aborted=aborted)
|
|
|
|
# Cleanup
|
|
state.update_progress_pane()
|
|
for pane in state.panes['SMART']:
|
|
tmux.kill_pane(pane)
|
|
state.panes.pop('SMART', None)
|
|
|
|
# Done
|
|
if aborted:
|
|
raise std.GenericAbort('Aborted')
|
|
|
|
|
|
def disk_surface_scan(state, test_objects) -> None:
|
|
"""Read-only disk surface scan using badblocks."""
|
|
LOG.info('Disk Surface Scan (badblocks)')
|
|
aborted = False
|
|
threads = []
|
|
state.panes['badblocks'] = []
|
|
|
|
# Update panes
|
|
state.update_top_pane(
|
|
f'Disk Surface Scan{"s" if len(test_objects) > 1 else ""}',
|
|
)
|
|
std.print_info(
|
|
f'Starting disk surface scan{"s" if len(test_objects) > 1 else ""}',
|
|
)
|
|
for disk in state.disks:
|
|
failed_attributes = [
|
|
line for line in hw_smart.generate_attribute_report(disk) if 'failed' in line
|
|
]
|
|
if failed_attributes:
|
|
size_str = std.bytes_to_string(disk.size, use_binary=False)
|
|
std.print_colored(
|
|
['[', disk.path.name, ' ', size_str, ']'],
|
|
[None, 'BLUE', None, 'CYAN', None],
|
|
sep='',
|
|
)
|
|
std.print_report(failed_attributes)
|
|
std.print_standard('')
|
|
|
|
# Run surface scans
|
|
for test in reversed([test for test in test_objects if not test.disabled]):
|
|
|
|
# Start thread
|
|
test_log = f'{state.log_dir}/{test.dev.path.name}_badblocks.log'
|
|
threads.append(exe.start_thread(
|
|
hw_surface_scan.run_scan, args=(test, test_log),
|
|
))
|
|
|
|
# Show progress
|
|
if threads[-1].is_alive():
|
|
state.panes['badblocks'].append(
|
|
tmux.split_window(
|
|
lines=5,
|
|
vertical=True,
|
|
watch_cmd='tail',
|
|
watch_file=test_log,
|
|
),
|
|
)
|
|
|
|
# Wait for all tests to complete
|
|
try:
|
|
while True:
|
|
if any(t.is_alive() for t in threads):
|
|
state.update_progress_pane()
|
|
std.sleep(5)
|
|
else:
|
|
break
|
|
except KeyboardInterrupt:
|
|
aborted = True
|
|
std.sleep(0.5)
|
|
# Handle aborts
|
|
for test in test_objects:
|
|
if not (test.disabled or test.passed or test.failed):
|
|
test.set_status('Aborted')
|
|
test.report.append(std.color_string(' Aborted', 'YELLOW'))
|
|
|
|
# Cleanup
|
|
state.update_progress_pane()
|
|
for pane in state.panes['badblocks']:
|
|
tmux.kill_pane(pane)
|
|
state.panes.pop('badblocks', None)
|
|
|
|
# Done
|
|
if aborted:
|
|
raise std.GenericAbort('Aborted')
|
|
|
|
|
|
def main() -> None:
|
|
# pylint: disable=too-many-branches
|
|
"""Main function for hardware diagnostics."""
|
|
args = docopt(DOCSTRING)
|
|
log.update_log_path(dest_name='Hardware-Diagnostics', timestamp=True)
|
|
|
|
# Safety check
|
|
if 'TMUX' not in os.environ:
|
|
LOG.error('tmux session not found')
|
|
raise RuntimeError('tmux session not found')
|
|
|
|
# Init
|
|
atexit.register(tmux.kill_all_panes)
|
|
menu = build_menu(cli_mode=args['--cli'], quick_mode=args['--quick'])
|
|
state = State()
|
|
|
|
# Quick Mode
|
|
if args['--quick']:
|
|
run_diags(state, menu, quick_mode=True)
|
|
return
|
|
|
|
# Show menu
|
|
while True:
|
|
action = None
|
|
selection = menu.advanced_select()
|
|
|
|
# Set action
|
|
if 'Audio Test' in selection:
|
|
action = audio_test
|
|
elif 'Keyboard Test' in selection:
|
|
action = keyboard_test
|
|
elif 'Network Test' in selection:
|
|
action = network_test
|
|
elif 'Clock Sync' in selection:
|
|
action = sync_clock
|
|
|
|
# Run simple test
|
|
if action:
|
|
state.update_top_pane(selection[0])
|
|
try:
|
|
action()
|
|
except KeyboardInterrupt:
|
|
std.print_warning('Aborted.')
|
|
std.print_standard('')
|
|
std.pause('Press Enter to return to main menu...')
|
|
if 'Clock Sync' in selection:
|
|
state.update_clock()
|
|
|
|
# Secrets
|
|
if 'Matrix' in selection:
|
|
screensaver('matrix')
|
|
elif 'Tubes' in selection:
|
|
# Tubes ≈≈ Pipes?
|
|
screensaver('pipes')
|
|
|
|
# Quit
|
|
if 'Reboot' in selection:
|
|
cmd = ['/usr/local/bin/wk-power-command', 'reboot']
|
|
exe.run_program(cmd, check=False)
|
|
elif 'Power Off' in selection:
|
|
cmd = ['/usr/local/bin/wk-power-command', 'poweroff']
|
|
exe.run_program(cmd, check=False)
|
|
elif 'Quit' in selection:
|
|
break
|
|
|
|
# Start diagnostics
|
|
if 'Start' in selection:
|
|
run_diags(state, menu, quick_mode=False)
|
|
|
|
# Reset top pane
|
|
state.update_top_pane('Main Menu')
|
|
|
|
|
|
def print_countdown(proc, seconds) -> None:
|
|
"""Print countdown to screen while proc is alive."""
|
|
for i in range(seconds):
|
|
sec_left = (seconds - i) % 60
|
|
min_left = int((seconds - i) / 60)
|
|
|
|
out_str = '\r '
|
|
if min_left:
|
|
out_str += f'{min_left} minute{"s" if min_left != 1 else ""}, '
|
|
out_str += f'{sec_left} second{"s" if sec_left != 1 else ""}'
|
|
out_str += ' remaining'
|
|
|
|
print(f'{out_str:<42}', end='', flush=True)
|
|
try:
|
|
proc.wait(1)
|
|
except subprocess.TimeoutExpired:
|
|
# proc still going, continue
|
|
pass
|
|
if ((hasattr(proc, 'poll') and proc.poll() is not None)
|
|
or (hasattr(proc, 'is_running') and not proc.is_running())):
|
|
# proc exited, stop countdown
|
|
break
|
|
|
|
# Done
|
|
print('')
|
|
|
|
|
|
def run_diags(state, menu, quick_mode=False) -> None:
|
|
"""Run selected diagnostics."""
|
|
aborted = False
|
|
atexit.register(state.save_debug_reports)
|
|
state.init_diags(menu)
|
|
|
|
# Just return if no tests were selected
|
|
if not state.test_groups:
|
|
std.print_warning('No tests selected?')
|
|
std.pause()
|
|
return
|
|
|
|
# Run tests
|
|
for group in state.test_groups:
|
|
|
|
# Run test(s)
|
|
function = group.function
|
|
args = [group.test_objects]
|
|
if group.name == 'Disk I/O Benchmark':
|
|
args.append(menu.toggles['Skip USB Benchmarks']['Selected'])
|
|
std.clear_screen()
|
|
try:
|
|
function(state, *args)
|
|
except (KeyboardInterrupt, std.GenericAbort):
|
|
aborted = True
|
|
state.abort_testing()
|
|
state.update_progress_pane()
|
|
break
|
|
|
|
# Run safety checks
|
|
if group.name.startswith('Disk'):
|
|
state.disk_safety_checks(
|
|
wait_for_self_tests=group.name != 'Disk Attributes',
|
|
)
|
|
|
|
# Handle aborts
|
|
if aborted:
|
|
for group in state.test_groups:
|
|
for test in group.test_objects:
|
|
if test.status == 'Pending':
|
|
test.set_status('Aborted')
|
|
|
|
# Show results
|
|
show_results(state)
|
|
|
|
# Done
|
|
state.save_debug_reports()
|
|
atexit.unregister(state.save_debug_reports)
|
|
if quick_mode:
|
|
std.pause('Press Enter to exit...')
|
|
else:
|
|
std.pause('Press Enter to return to main menu...')
|
|
|
|
|
|
def show_results(state) -> None:
|
|
"""Show test results by device."""
|
|
std.sleep(0.5)
|
|
std.clear_screen()
|
|
state.update_top_pane('Results')
|
|
|
|
# CPU Tests
|
|
cpu_tests_enabled = [
|
|
group.name for group in state.test_groups if 'CPU' in group.name
|
|
]
|
|
if cpu_tests_enabled:
|
|
std.print_success('CPU:')
|
|
std.print_report(state.system.generate_report())
|
|
std.print_standard(' ')
|
|
|
|
# Disk Tests
|
|
disk_tests_enabled = [
|
|
group.name for group in state.test_groups if 'Disk' in group.name
|
|
]
|
|
if disk_tests_enabled:
|
|
std.print_success(f'Disk{"s" if len(state.disks) > 1 else ""}:')
|
|
for disk in state.disks:
|
|
std.print_report(disk.generate_report())
|
|
std.print_standard(' ')
|
|
if not state.disks:
|
|
std.print_warning('No devices')
|
|
std.print_standard(' ')
|
|
|
|
|
|
def sync_clock() -> None:
|
|
"""Sync clock under macOS using sntp."""
|
|
cmd = ['sudo', 'sntp', '-Ss', 'us.pool.ntp.org']
|
|
proc = exe.run_program(cmd, check=False)
|
|
if proc.returncode:
|
|
# Assuming we're running under an older version of macOS
|
|
cmd[2] = '-s'
|
|
exe.run_program(cmd, check=False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
print("This file is not meant to be called directly.")
|