Added get_disks()
* This calls either get_disks_linux() or get_disks_macos()
This commit is contained in:
parent
4e6b2cd4da
commit
1054794af3
1 changed files with 85 additions and 0 deletions
|
|
@ -6,6 +6,8 @@ import logging
|
|||
import os
|
||||
import pathlib
|
||||
import platform
|
||||
import plistlib
|
||||
import re
|
||||
import signal
|
||||
import time
|
||||
|
||||
|
|
@ -13,6 +15,7 @@ from collections import OrderedDict
|
|||
from docopt import docopt
|
||||
|
||||
from wk import cfg, exe, log, net, std, tmux
|
||||
from wk.hw import obj as hw_obj
|
||||
|
||||
|
||||
# atexit functions
|
||||
|
|
@ -63,6 +66,10 @@ MENU_SETS = {
|
|||
MENU_TOGGLES = (
|
||||
'Skip USB Benchmarks',
|
||||
)
|
||||
WK_LABEL_REGEX = re.compile(
|
||||
fr'{cfg.main.KIT_NAME_SHORT}_(LINUX|UFD)',
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
# Classes
|
||||
|
|
@ -153,6 +160,12 @@ class State():
|
|||
)
|
||||
std.print_info('Starting Hardware Diagnostics')
|
||||
|
||||
# Add CPU
|
||||
self.cpu = hw_obj.CpuRam()
|
||||
|
||||
# Add disks
|
||||
self.disks = get_disks()
|
||||
|
||||
def init_tmux(self):
|
||||
"""Initialize tmux layout."""
|
||||
tmux.kill_all_panes()
|
||||
|
|
@ -286,6 +299,78 @@ def disk_surface_scan():
|
|||
std.print_warning('TODO: bb')
|
||||
|
||||
|
||||
def get_disks():
|
||||
"""Get disks using OS-specific methods, returns list."""
|
||||
disks = []
|
||||
if platform.system() == 'Darwin':
|
||||
disks = get_disks_macos()
|
||||
elif platform.system() == 'Linux':
|
||||
disks = get_disks_linux()
|
||||
|
||||
# Done
|
||||
return disks
|
||||
|
||||
|
||||
def get_disks_linux():
|
||||
"""Get disks via lsblk, returns list."""
|
||||
cmd = ['lsblk', '--json', '--nodeps', '--paths']
|
||||
disks = []
|
||||
|
||||
# Add valid disks
|
||||
json_data = exe.get_json_from_command(cmd)
|
||||
for disk in json_data.get('blockdevices', []):
|
||||
disk_obj = hw_obj.Disk(disk['name'])
|
||||
skip = False
|
||||
|
||||
# Skip loopback devices, optical devices, etc
|
||||
if disk_obj.details['type'] != 'disk':
|
||||
skip = True
|
||||
|
||||
# Skip WK disks
|
||||
for label in disk_obj.get_labels():
|
||||
if WK_LABEL_REGEX.search(label):
|
||||
skip = True
|
||||
|
||||
# Add disk
|
||||
if not skip:
|
||||
disks.append(disk_obj)
|
||||
|
||||
# Done
|
||||
return disks
|
||||
|
||||
|
||||
def get_disks_macos():
|
||||
"""Get disks via diskutil, returns list."""
|
||||
cmd = ['diskutil', 'list', '-plist', 'physical']
|
||||
disks = []
|
||||
|
||||
# Get info from diskutil
|
||||
proc = exe.run_program(cmd, encoding=None, errors=None)
|
||||
try:
|
||||
plist_data = plistlib.loads(proc.stdout)
|
||||
except (TypeError, ValueError):
|
||||
# Invalid / corrupt plist data? return empty list to avoid crash
|
||||
LOG.error('Failed to get diskutil list')
|
||||
return disks
|
||||
|
||||
# Add valid disks
|
||||
for disk in plist_data['WholeDisks']:
|
||||
disk_obj = hw_obj.Disk(disk)
|
||||
skip = False
|
||||
|
||||
# Skip WK disks
|
||||
for label in disk_obj.get_labels():
|
||||
if WK_LABEL_REGEX.search(label):
|
||||
skip = True
|
||||
|
||||
# Add disk
|
||||
if not skip:
|
||||
disks.append(disk_obj)
|
||||
|
||||
# Done
|
||||
return disks
|
||||
|
||||
|
||||
def keyboard_test():
|
||||
"""Test keyboard using xev."""
|
||||
LOG.info('Keyboard Test (xev)')
|
||||
|
|
|
|||
Loading…
Reference in a new issue