peterlu14
482 lines
12 KiB
Python
482 lines
12 KiB
Python
import re
|
|
from pathlib import Path
|
|
from subprocess import check_output, CalledProcessError
|
|
from typing import Optional, List, Union
|
|
|
|
from biopro.util.cli import *
|
|
from biopro.util.sys import system_sudo, system_call
|
|
from biopro.util.text import list_padding
|
|
|
|
|
|
class Partition:
|
|
_BLOCK_INTERNAL: Optional[str] = None
|
|
|
|
SUPPORT_FSTYPE = ('vfat', 'ext4')
|
|
|
|
__slots__ = ('_fs_path', '_fs_type', '_path', '_usage', '_size', '_label', '_bind')
|
|
|
|
def __init__(self, fs_path: str):
|
|
self._fs_path = fs_path
|
|
self._fs_type = ''
|
|
self._path: Optional[str] = None
|
|
self._usage: Optional[int] = None
|
|
self._size: int = 0
|
|
self._label: Optional[str] = None
|
|
self._bind = False
|
|
|
|
@property
|
|
def fs_path(self) -> str:
|
|
return self._fs_path
|
|
|
|
@property
|
|
def fs_type(self) -> str:
|
|
return self._fs_type
|
|
|
|
@property
|
|
def fs_size(self) -> int:
|
|
return self._size
|
|
|
|
@property
|
|
def fs_used(self) -> int:
|
|
return self._usage
|
|
|
|
@property
|
|
def fs_avail(self) -> int:
|
|
return self._size - self._usage
|
|
|
|
@property
|
|
def fs_label(self) -> Optional[str]:
|
|
return self._label
|
|
|
|
@property
|
|
def mount_path(self) -> Optional[str]:
|
|
return self._path
|
|
|
|
@property
|
|
def is_external_storage(self) -> bool:
|
|
self._init_attribute()
|
|
|
|
return not self._fs_path.startswith(self._BLOCK_INTERNAL)
|
|
|
|
def sync(self):
|
|
self._sync_lsblk()
|
|
self._sync_df()
|
|
|
|
def _sync_df(self):
|
|
if self._path is None:
|
|
return
|
|
|
|
elif self._path == '[SWAP]':
|
|
self._fs_type = 'SWAP'
|
|
return
|
|
|
|
output = system_call('df', '--output=fstype,size,used,target', self._path)
|
|
|
|
part: List[str] = output.split('\n')
|
|
part = part[-2].split(' ')
|
|
part = list(filter(len, part))
|
|
|
|
fs_type = part[0]
|
|
|
|
if fs_type != 'devtmpfs':
|
|
self._fs_type = fs_type
|
|
self._size = int(part[1]) * 1024 # 1K to 1024B
|
|
self._usage = int(part[2])
|
|
|
|
try:
|
|
self._path = part[3]
|
|
except IndexError:
|
|
pass
|
|
|
|
def _sync_lsblk(self):
|
|
if self._bind:
|
|
return
|
|
|
|
# noinspection SpellCheckingInspection
|
|
output = system_call('lsblk', '-bfnpr', '-o', 'NAME,FSTYPE,SIZE,LABEL,MOUNTPOINT', self._fs_path)
|
|
|
|
part: List[str] = output.split('\n')
|
|
part: List[str] = part[0].split(' ')
|
|
|
|
self._fs_type = part[1]
|
|
self._size = int(part[2]) # Bytes
|
|
self._label = part[3]
|
|
self._usage = None
|
|
self._path = None
|
|
|
|
try:
|
|
if len(part[4]) > 0:
|
|
self._path = part[4]
|
|
except IndexError:
|
|
pass
|
|
|
|
self._label = self._label.replace('\\x20', ' ')
|
|
|
|
def mount(self, path: Union[str, Path]):
|
|
self.sync()
|
|
|
|
# mount checking
|
|
if self._path is not None:
|
|
raise OSError('%s has mount at %s' % (self._fs_path, self._path))
|
|
|
|
# mount path checking
|
|
if isinstance(path, str):
|
|
path = Path(path)
|
|
|
|
if not path.exists() or not path.is_dir():
|
|
raise NotADirectoryError(path)
|
|
|
|
# mount command
|
|
mount_cmd = ['mount']
|
|
|
|
# mount command options for type of file system
|
|
fs_type = self._fs_type
|
|
|
|
if fs_type == 'ext4':
|
|
mount_cmd.append('-o')
|
|
mount_cmd.append('uid=pi,gid=pi')
|
|
|
|
elif fs_type == 'vfat':
|
|
mount_cmd.append('-o')
|
|
mount_cmd.append('uid=pi,gid=pi,umask=007')
|
|
|
|
elif fs_type == '':
|
|
raise RuntimeError('unknown file system : ' + fs_type)
|
|
|
|
else:
|
|
raise RuntimeError('unsupported file system : ' + fs_type)
|
|
|
|
# mount command arguments
|
|
mount_cmd.append(self._fs_path)
|
|
mount_cmd.append(str(path))
|
|
|
|
# sudo
|
|
try:
|
|
system_sudo(*mount_cmd)
|
|
except CalledProcessError as e:
|
|
raise OSError('cannot umount ' + self._fs_path + '. sudo error') from e
|
|
|
|
finally:
|
|
self.sync()
|
|
|
|
def mount_with_label(self, root: Union[str, Path], label: Optional[str] = None) -> Path:
|
|
self.sync()
|
|
|
|
# mount checking
|
|
if self._path is not None:
|
|
raise OSError('%s has mount at %s' % (self._fs_path, self._path))
|
|
|
|
# root path
|
|
if isinstance(root, str):
|
|
root = Path(root)
|
|
|
|
if not root.exists() or not root.is_dir():
|
|
raise NotADirectoryError(root)
|
|
|
|
# partition label
|
|
if label is None:
|
|
label = self._label
|
|
|
|
if label is None:
|
|
label = Path(self._fs_path).name
|
|
|
|
else:
|
|
label = label.replace(' ', '_')
|
|
|
|
# mount path checking
|
|
path = root / label
|
|
|
|
exists = path.exists()
|
|
|
|
if not path.exists():
|
|
try:
|
|
path.mkdir(parents=True)
|
|
except OSError:
|
|
pass
|
|
|
|
if not path.exists():
|
|
try:
|
|
system_sudo('mkdir', '-p', str(path))
|
|
except CalledProcessError:
|
|
pass
|
|
|
|
if not path.exists():
|
|
raise RuntimeError('cannot create directory at : ' + str(path))
|
|
|
|
# mounting
|
|
try:
|
|
self.mount(path)
|
|
|
|
return path
|
|
except:
|
|
# remove directory
|
|
if not exists:
|
|
path.rmdir()
|
|
|
|
raise
|
|
|
|
def umount(self):
|
|
self.sync()
|
|
|
|
if self._path is None:
|
|
return
|
|
|
|
try:
|
|
system_sudo('umount', '-Rf', self._path)
|
|
|
|
except CalledProcessError as e:
|
|
raise OSError('cannot umount ' + self._fs_path + ' from ' + self._path) from e
|
|
|
|
finally:
|
|
self.sync()
|
|
|
|
def __hash__(self):
|
|
return hash(self.fs_path)
|
|
|
|
def __eq__(self, other):
|
|
if self is other:
|
|
return True
|
|
|
|
elif other is None:
|
|
return False
|
|
|
|
elif isinstance(other, Partition):
|
|
return self.fs_path == other.fs_path
|
|
else:
|
|
return False
|
|
|
|
def __str__(self):
|
|
if len(self._fs_type) == 0:
|
|
# not sync
|
|
return '%s (<not-sync>) [??/??] ??' % self._fs_path
|
|
else:
|
|
return '%s (%s) [%s/%s] %s' % (self._fs_path,
|
|
self._fs_type,
|
|
size_to_human_readable(self._usage) if self._usage is not None else '',
|
|
size_to_human_readable(self._size) if self._size is not None else '',
|
|
self._path)
|
|
|
|
def __repr__(self) -> str:
|
|
return str(self)
|
|
|
|
@classmethod
|
|
def bind(cls, from_path: Path, target_path: Path) -> 'Partition':
|
|
if not from_path.exists():
|
|
raise FileNotFoundError(str(from_path))
|
|
|
|
if not target_path.exists():
|
|
raise FileNotFoundError(str(target_path))
|
|
|
|
try:
|
|
system_sudo('mount', '--bind', str(from_path), str(target_path))
|
|
except CalledProcessError as e:
|
|
raise OSError('cannot umount ' + str(from_path) + '. sudo error') from e
|
|
|
|
part = Partition(str(from_path))
|
|
part._bind = True
|
|
part._path = str(target_path)
|
|
|
|
return part
|
|
|
|
@classmethod
|
|
def get_block_device(cls, path: str) -> Optional['Partition']:
|
|
for p in Partition.list_block_device():
|
|
if p.fs_path == path:
|
|
p.sync()
|
|
return p
|
|
|
|
return None
|
|
|
|
@classmethod
|
|
def get_mount_point(cls, path: Union[str, Path]) -> Optional['Partition']:
|
|
if isinstance(path, str):
|
|
path = Path(path)
|
|
|
|
path = path.absolute()
|
|
|
|
def _key(_p: Partition):
|
|
if _p.mount_path is None:
|
|
return 0
|
|
elif _p.mount_path == '/':
|
|
return 0
|
|
else:
|
|
return _p.mount_path.count('/')
|
|
|
|
for p in sorted(cls.list_block_device(sync=True), key=_key, reverse=True):
|
|
mp = p.mount_path
|
|
|
|
if mp is not None:
|
|
try:
|
|
path.relative_to(mp)
|
|
return p
|
|
except ValueError:
|
|
pass
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def list_block_device(sync: bool = False) -> List['Partition']:
|
|
# noinspection SpellCheckingInspection
|
|
output: bytes = check_output(['lsblk', '-bfnpr', '-o', 'NAME,FSTYPE,MODE'])
|
|
output: str = output.decode()
|
|
|
|
ret = []
|
|
|
|
for line in output.split('\n'):
|
|
part = line.split(' ')
|
|
|
|
if len(part) == 3 and len(part[1]) > 0 and part[2] == 'brw-rw----':
|
|
p = Partition(part[0])
|
|
ret.append(p)
|
|
|
|
if sync:
|
|
p.sync()
|
|
|
|
return ret
|
|
|
|
@classmethod
|
|
def _init_attribute(cls):
|
|
if cls._BLOCK_INTERNAL is None:
|
|
p = cls.get_mount_point(Path.cwd())
|
|
|
|
if p is None:
|
|
raise RuntimeError('cannot find current mount at which partition')
|
|
|
|
name = p._fs_path
|
|
cls._BLOCK_INTERNAL = re.sub('\\d+$', '', name)
|
|
|
|
@classmethod
|
|
def print_lsblk(cls):
|
|
t1 = ['Path']
|
|
t2 = ['Type']
|
|
t3 = ['Size']
|
|
t4 = ['Used']
|
|
t5 = ['Avail']
|
|
t6 = ['Used%']
|
|
t7 = ['Mount']
|
|
t8 = ['Label']
|
|
|
|
for p in cls.list_block_device(sync=True):
|
|
t1.append(p.fs_path)
|
|
t2.append(p.fs_type)
|
|
t3.append(size_to_human_readable(p.fs_size))
|
|
|
|
u = p.fs_used
|
|
if u is not None:
|
|
t4.append(size_to_human_readable(p.fs_used))
|
|
t5.append(size_to_human_readable(p.fs_avail))
|
|
t6.append(str(round(100 * p.fs_used / p.fs_size, 1)) + '%')
|
|
else:
|
|
t4.append('')
|
|
t5.append('')
|
|
t6.append('')
|
|
|
|
t7.append(p.mount_path if p.mount_path is not None else '')
|
|
t8.append(p.fs_label if p.fs_label is not None else '')
|
|
|
|
list_padding(t1, t2, align_right=True)
|
|
list_padding(t1, t3, align_right=True, split=' ')
|
|
list_padding(t1, t4, align_right=True, split=' ')
|
|
list_padding(t1, t5, align_right=True, split=' ')
|
|
list_padding(t1, t6, align_right=True, split=' ')
|
|
list_padding(t1, t7, align_right=False, split=' ')
|
|
list_padding(t1, t8, split=' ')
|
|
|
|
for line in t1:
|
|
print(line)
|
|
|
|
|
|
def size_to_human_readable(size: int) -> str:
|
|
size = float(size)
|
|
unit = 'B'
|
|
|
|
if size >= 1024:
|
|
size /= 1024
|
|
unit = 'K'
|
|
|
|
if size >= 1024:
|
|
size /= 1024
|
|
unit = 'M'
|
|
|
|
if size >= 1024:
|
|
size /= 1024
|
|
unit = 'G'
|
|
|
|
if size >= 1024:
|
|
size /= 1024
|
|
unit = 'T'
|
|
|
|
size = round(size, 3)
|
|
|
|
if size < 10:
|
|
return '%.3f%s' % (size, unit)
|
|
|
|
size = round(size, 2)
|
|
|
|
if size < 100:
|
|
return '%.2f%s' % (size, unit)
|
|
|
|
else:
|
|
return '%.1f%s' % (size, unit)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class Main(CliMain):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document"""
|
|
self.print_help()
|
|
|
|
@cli_command('lsblk')
|
|
def _list(self, act, argv: List[str]):
|
|
"""list block device"""
|
|
|
|
Partition.print_lsblk()
|
|
|
|
@cli_command('mount-point', value='PATH')
|
|
def _mount_point(self, act, argv: List[str]):
|
|
a = Arguments(argv)
|
|
p = a.required('PATH')
|
|
partition = Partition.get_mount_point(p)
|
|
if partition is None:
|
|
print('')
|
|
else:
|
|
print(partition.fs_path)
|
|
|
|
@cli_command('mount', value=('DEVICE', 'PATH'))
|
|
def _mount(self, act, argv: List[str]):
|
|
"""mount"""
|
|
|
|
a = Arguments(argv)
|
|
device, path = a.required('DEVICE', 'PATH')
|
|
|
|
partition = Partition.get_block_device(device)
|
|
|
|
if partition is None:
|
|
raise RuntimeError('device not found', device)
|
|
|
|
partition.mount(path)
|
|
|
|
@cli_command('umount', value='DEVICE')
|
|
def _umount(self, act, argv: List[str]):
|
|
"""umount"""
|
|
a = Arguments(argv)
|
|
device = a.required('DEVICE')
|
|
|
|
partition = Partition.get_block_device(device)
|
|
|
|
if partition is None:
|
|
raise RuntimeError('device not found', device)
|
|
|
|
partition.umount()
|
|
|
|
def run(self):
|
|
self._help('')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|