#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Name: imager.py # Purpose: Module to write ISO image to selected USB disk. Uses dd as backend. # Authors: Sundar # Licence: This file is a part of multibootusb package. You can redistribute it or modify # under the terms of GNU General Public License, v.2 or above # WARNING : Any boot-able USB made using this module will destroy data stored on USB disk. import os import subprocess import collections import platform import signal from PyQt5 import QtGui from PyQt5 import QtWidgets from PyQt5 import QtCore from .gui.ui_multibootusb import Ui_MainWindow from .gen import * from . import iso from . import usb from . import config from . import progressbar if platform.system() == "Windows": import win32com.client def dd_linux(): import time input = "if=" + config.image_path in_file_size = float(os.path.getsize(config.image_path)) output = "of=" + config.usb_disk os.system("umount " + config.usb_disk + "1") command = ['dd', input, output, "bs=1M", "oflag=sync"] log("Executing ==> " + " ".join(command)) dd_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False) # bar = progressbar.ProgressBar(redirect_stdout=True) pbar = progressbar.ProgressBar( maxval=100, widgets=[ ' ', progressbar.widgets.Bar(marker='=', left='[', right=']'), ' ', progressbar.widgets.Percentage() ] ).start() while dd_process.poll() is None: time.sleep(0.1) # If this time delay is not given, the Popen does not execute the actual command dd_process.send_signal(signal.SIGUSR1) dd_process.stderr.flush() while True: time.sleep(0.1) out_error = dd_process.stderr.readline().decode() if out_error: if 'bytes' in out_error: copied = int(out_error.split(' ', 1)[0]) config.imager_percentage = round((float(copied) / float(in_file_size) * 100)) pbar.update(config.imager_percentage) break if dd_process.poll() is not None: log("\nExecuting ==> sync") os.system("sync") log("ISO has been written to USB disk...") return def dd_win(): windd = resource_path(os.path.join("data", "tools", "dd", "dd.exe")) if os.path.exists(resource_path(os.path.join("data", "tools", "dd", "dd.exe"))): log("dd exist") input = "if=" + config.image_path in_file_size = float(os.path.getsize(config.image_path) / 1024 / 1024) output = "of=\\\.\\" + config.usb_disk command = [windd, input, output, "bs=1M", "--progress"] log("Executing ==> " + " ".join(command)) dd_process = subprocess.Popen(command, universal_newlines=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=False) while dd_process.poll() is None: for line in iter(dd_process.stderr.readline, ''): line = line.strip() if 'error' in line.lower() or 'invalid' in line.lower(): log("Error writing to disk...") break if line and line[-1] == 'M': copied = float(line.strip('M').replace(',', '')) config.imager_percentage = round((copied / float(in_file_size) * 100)) log("ISO has been written to USB disk...") return class Imager(QtWidgets.QMainWindow, Ui_MainWindow): """ Raw write to USB disk using dd. """ def __init__(self): QtWidgets.QMainWindow.__init__(self) self.ui = Ui_MainWindow() self.ui.setupUi(self) def add_iso_gui_label_text(self): """ Simple function to add text label to GUI widgets. :return: """ log("Testing ISO...") self.ui.imager_bootable.setVisible(True) if iso.is_bootable(config.image_path) is True: self.ui.imager_bootable.setText("Bootable ISO: Yes") log("ISO is bootable.") else: self.ui.imager_bootable.setText("Bootable ISO: No") log("ISO is not bootable.") if os.path.exists(config.image_path): log("Path " + config.image_path + " exists...") self.iso_size = str(round(os.path.getsize(config.image_path) / 1024 / 1024)) self.ui.imager_iso_size.setVisible(True) self.ui.imager_iso_size.setText("ISO Size: " + self.iso_size + " MB") log("ISO Size is " + self.iso_size + " MB") # def onImagerComboChange(self): # config.imager_usb_disk = str(self.ui.comboBox_2.currentText()) # if bool(config.imager_usb_disk): # self.ui.imager_disk_label.setText(self.imager_usb_detail(config.imager_usb_disk, partition=0).usb_type) # self.ui.imager_total_size.setText(usb.bytes2human(int(self.imager_usb_detail(config.imager_usb_disk, partition=0).total_size))) # # if platform.system() == "Linux": # self.ui.label_imager_uuid.setText("Disk Model:") # self.ui.imager_uuid.setText(str(self.imager_usb_detail(config.imager_usb_disk, partition=0).model)) # else: # self.ui.imager_uuid.setText(self.imager_usb_detail(config.imager_usb_disk, partition=0).model) def imager_list_usb(self, partition=1): """ Function to detect whole USB disk. It uses lsblk package on Linux. :param partition: What to return. By default partition is set. :return: USB disk/partition as list """ disk = [] if platform.system() == "Linux": output = subprocess.check_output("lsblk -i", shell=True) if not partition == 1: for line in output.splitlines(): line = line.split() if (line[2].strip()) == b'1' and (line[5].strip()) == b'disk': disk.append(str("/dev/" + str(line[0].strip().decode()))) elif partition == 1: for line in output.splitlines(): line = line.split() if (line[2].strip()) == b'1' and line[5].strip() == b'part': disk.append(str("/dev/" + str(line[0].strip()[2:]))) else: if partition == 1 or not partition == 1: oFS = win32com.client.Dispatch("Scripting.FileSystemObject") oDrives = oFS.Drives for drive in oDrives: if drive.DriveType == 1 and drive.IsReady: disk.append(drive) return disk def imager_usb_detail(self, usb_disk, partition=1): """ Function to detect details of USB disk using lsblk :param usb_disk: path to usb disk :param partition: by default partition is set (but yet to code for it) :return: details of size, type and model as tuples """ _ntuple_diskusage = collections.namedtuple('usage', 'total_size usb_type model') if platform.system() == "Linux": output = subprocess.check_output("lsblk -ib " + usb_disk, shell=True) for line in output.splitlines(): line = line.split() if not partition == 1: if line[2].strip() == b'1' and line[5].strip() == b'disk': total_size = line[3] if not total_size: total_size = "Unknown" usb_type = "Removable" model = subprocess.check_output("lsblk -in -f -o MODEL " + usb_disk, shell=True).decode().strip() if not model: model = "Unknown" else: try: selected_usb_part = str(usb_disk) oFS = win32com.client.Dispatch("Scripting.FileSystemObject") d = oFS.GetDrive(oFS.GetDriveName(oFS.GetAbsolutePathName(selected_usb_part))) selected_usb_device = d.DriveLetter label = (d.VolumeName).strip() if not label.strip(): label = "No label." total_size = d.TotalSize usb_type = "Removable" model = label except: log("Error detecting USB details.") return _ntuple_diskusage(total_size, usb_type, model) def get_usb_size(self, usb_disk): """ Function to detect USB disk space. Useful but not used in multibootusb as of now. :param usb_disk: USB disk like "/dev/sdb" :return: Size of the disk as integer """ if platform.system() == "Linux": cat_output = subprocess.check_output("cat /proc/partitions | grep " + usb_disk[5:], shell=True) usb_size = int(cat_output.split()[2]) * 1024 # log(usb_size) return usb_size else: usb_size = self.usb.disk_usage(self.usb.get_usb(usb_disk).mount).total return usb_size