diff options
| -rw-r--r-- | mpm/python/usrp_mpm/sys_utils/mount.py | 96 |
1 files changed, 96 insertions, 0 deletions
diff --git a/mpm/python/usrp_mpm/sys_utils/mount.py b/mpm/python/usrp_mpm/sys_utils/mount.py new file mode 100644 index 000000000..2e32fa014 --- /dev/null +++ b/mpm/python/usrp_mpm/sys_utils/mount.py @@ -0,0 +1,96 @@ +# +# Copyright 2020 Ettus Research, a National Instruments Brand +# +# SPDX-License-Identifier: GPL-3.0-or-later +# +""" +Utilities for creating a mount point +""" + +import subprocess +import os +from usrp_mpm.mpmlog import get_logger + +class Mount(): + """ + Class for creating a mount point + """ + + def __init__(self, devicepath, mountpoint, options=None, log=None): + assert isinstance(devicepath, str) + assert isinstance(mountpoint, str) + assert isinstance(options, list) + self.devicepath = devicepath + self.mountpoint = mountpoint + self.options = options + if log is None: + self.log = get_logger("Mount") + else: + self.log = log.getChild("Mount") + self.log.trace("Early initialization: devicepath={}, mountpoint={}, options={}".format( + devicepath, mountpoint, options)) + + def ismounted(self): + """ + Returns true if the mount point is mounted + """ + assert self.devicepath is not None + collection = [line.split()[0] for line in open("/etc/mtab") + if line.split()[0] == self.devicepath] + mounted = len(collection) != 0 + return mounted + + def get_mount_point(self): + """ + returns the mount point (None when not mounted) + """ + if not self.ismounted(): + return None + return self.mountpoint + + def prepare_mountpoint(self): + """ + Creates the mount point directory (if not already existing) + """ + if not os.path.exists(self.mountpoint): + os.makedirs(self.mountpoint) + return True + + def delete_mountpoint(self): + """ + Deletes the mount point directory + """ + os.removedirs(self.mountpoint) + return True + + def mount(self): + """ + Mounts the mount point + """ + if self.ismounted(): + self.log.warning("{} was already mounted".format(self.mountpoint)) + return True + self.prepare_mountpoint() + self.log.debug("Mounting {}".format(self.mountpoint)) + cmd = ['mount'] + if self.options: + cmd.extend(self.options) + cmd.append(self.devicepath) + cmd.append(self.mountpoint) + proc = subprocess.run(cmd, check=True) + self.log.trace(proc) + return True + + def unmount(self): + """ + Unmounts the mount point + """ + if not self.ismounted(): + self.log.warning("{} was not mounted".format(self.mountpoint)) + return True + self.log.debug("Unmounting {}".format(self.mountpoint)) + cmd = ['umount', self.mountpoint] + proc = subprocess.run(cmd, check=True) + self.log.trace(proc) + self.delete_mountpoint() + return True |