# usb_writer_macos.py
import subprocess
import os
import time
import shutil # For checking command existence
import plistlib # For parsing diskutil list -plist output
class USBWriterMacOS:
def __init__(self, device: str, opencore_qcow2_path: str, macos_qcow2_path: str,
progress_callback=None, enhance_plist_enabled: bool = False, target_macos_version: str = ""): # New args
self.device = device # Should be like /dev/diskX
self.opencore_qcow2_path = opencore_qcow2_path
self.macos_qcow2_path = macos_qcow2_path
self.progress_callback = progress_callback
self.enhance_plist_enabled = enhance_plist_enabled # Store
self.target_macos_version = target_macos_version # Store
pid = os.getpid()
self.opencore_raw_path = f"opencore_temp_{pid}.raw"
self.macos_raw_path = f"macos_main_temp_{pid}.raw"
self.temp_opencore_mount = f"/tmp/opencore_efi_temp_skyscope_{pid}"
self.temp_usb_esp_mount = f"/tmp/usb_esp_temp_skyscope_{pid}"
self.temp_macos_source_mount = f"/tmp/macos_source_temp_skyscope_{pid}"
self.temp_usb_macos_target_mount = f"/tmp/usb_macos_target_temp_skyscope_{pid}"
self.temp_files_to_clean = [self.opencore_raw_path, self.macos_raw_path]
self.temp_mount_points_to_clean = [
self.temp_opencore_mount, self.temp_usb_esp_mount,
self.temp_macos_source_mount, self.temp_usb_macos_target_mount
]
self.attached_raw_images_devices = [] # Store devices from hdiutil attach
def _report_progress(self, message: str):
print(message) # For standalone testing
if self.progress_callback:
self.progress_callback(message)
def _run_command(self, command: list[str], check=True, capture_output=False, timeout=None):
self._report_progress(f"Executing: {' '.join(command)}")
try:
process = subprocess.run(
command, check=check, capture_output=capture_output, text=True, timeout=timeout
)
if capture_output:
if process.stdout and process.stdout.strip():
self._report_progress(f"STDOUT: {process.stdout.strip()}")
if process.stderr and process.stderr.strip():
self._report_progress(f"STDERR: {process.stderr.strip()}")
return process
except subprocess.TimeoutExpired:
self._report_progress(f"Command {' '.join(command)} timed out after {timeout} seconds.")
raise
except subprocess.CalledProcessError as e:
self._report_progress(f"Error executing {' '.join(command)} (code {e.returncode}): {e.stderr or e.stdout or str(e)}")
raise
except FileNotFoundError:
self._report_progress(f"Error: Command '{command[0]}' not found. Is it installed and in PATH?")
raise
def _cleanup_temp_files(self):
self._report_progress("Cleaning up temporary image files...")
for f_path in self.temp_files_to_clean:
if os.path.exists(f_path):
try:
os.remove(f_path)
self._report_progress(f"Removed {f_path}")
except OSError as e:
self._report_progress(f"Error removing {f_path}: {e}")
def _unmount_path(self, mount_path_or_device, is_device=False, force=False):
target = mount_path_or_device
cmd_base = ["diskutil"]
action = "unmountDisk" if is_device else "unmount"
if force:
cmd = cmd_base + [action, "force", target]
else:
cmd = cmd_base + [action, target]
is_target_valid_for_unmount = (os.path.ismount(mount_path_or_device) and not is_device) or \
(is_device and os.path.exists(target))
if is_target_valid_for_unmount:
self._report_progress(f"Attempting to unmount {target} (Action: {action}, Force: {force})...")
self._run_command(cmd, check=False, timeout=30)
def _detach_raw_image_device(self, device_path):
if device_path and os.path.exists(device_path):
self._report_progress(f"Detaching raw image device {device_path}...")
try:
info_check = subprocess.run(["diskutil", "info", device_path], capture_output=True, text=True, check=False)
if info_check.returncode == 0:
self._run_command(["hdiutil", "detach", device_path, "-force"], check=False, timeout=30)
else:
self._report_progress(f"Device {device_path} appears invalid or already detached.")
except Exception as e:
self._report_progress(f"Exception while checking/detaching {device_path}: {e}")
def _cleanup_all_mounts_and_mappings(self):
self._report_progress("Cleaning up all temporary mounts and attached raw images...")
for mp in reversed(self.temp_mount_points_to_clean):
self._unmount_path(mp, force=True)
if os.path.exists(mp):
try: os.rmdir(mp)
except OSError as e: self._report_progress(f"Could not rmdir {mp}: {e}")
devices_to_detach = list(self.attached_raw_images_devices)
for dev_path in devices_to_detach:
self._detach_raw_image_device(dev_path)
self.attached_raw_images_devices = []
def check_dependencies(self):
self._report_progress("Checking dependencies (qemu-img, diskutil, hdiutil, rsync)...")
dependencies = ["qemu-img", "diskutil", "hdiutil", "rsync"]
missing_deps = []
for dep in dependencies:
if not shutil.which(dep):
missing_deps.append(dep)
if missing_deps:
msg = f"Missing dependencies: {', '.join(missing_deps)}. `qemu-img` might need to be installed (e.g., via Homebrew: `brew install qemu`). `diskutil`, `hdiutil`, `rsync` are usually standard on macOS."
self._report_progress(msg)
raise RuntimeError(msg)
self._report_progress("All critical dependencies found.")
return True
def _get_partition_device_id(self, parent_disk_id_str: str, partition_label_or_type: str) -> str | None:
"""Finds partition device ID by Volume Name or Content Hint."""
target_disk_id = parent_disk_id_str.replace("/dev/", "")
self._report_progress(f"Searching for partition '{partition_label_or_type}' on disk '{target_disk_id}'")
try:
result = self._run_command(["diskutil", "list", "-plist", target_disk_id], capture_output=True)
if not result.stdout:
self._report_progress(f"No stdout from diskutil list for {target_disk_id}")
return None
plist_data = plistlib.loads(result.stdout.encode('utf-8'))
all_disks_and_partitions = plist_data.get("AllDisksAndPartitions", [])
if not isinstance(all_disks_and_partitions, list):
if plist_data.get("DeviceIdentifier") == target_disk_id:
all_disks_and_partitions = [plist_data]
else:
all_disks_and_partitions = []
for disk_info_entry in all_disks_and_partitions:
current_disk_id_in_plist = disk_info_entry.get("DeviceIdentifier")
if current_disk_id_in_plist == target_disk_id:
for part_info in disk_info_entry.get("Partitions", []):
vol_name = part_info.get("VolumeName")
content_hint = part_info.get("Content")
device_id = part_info.get("DeviceIdentifier")
if device_id:
if vol_name and vol_name.strip().lower() == partition_label_or_type.strip().lower():
self._report_progress(f"Found partition by VolumeName: {vol_name} -> /dev/{device_id}")
return f"/dev/{device_id}"
if content_hint and content_hint.strip().lower() == partition_label_or_type.strip().lower():
self._report_progress(f"Found partition by Content type: {content_hint} -> /dev/{device_id}")
return f"/dev/{device_id}"
self._report_progress(f"Partition '{partition_label_or_type}' not found on disk '{target_disk_id}'.")
return None
except Exception as e:
self._report_progress(f"Error parsing 'diskutil list -plist {target_disk_id}': {e}")
return None
def format_and_write(self) -> bool:
try:
self.check_dependencies()
self._cleanup_all_mounts_and_mappings()
for mp in self.temp_mount_points_to_clean:
os.makedirs(mp, exist_ok=True)
self._report_progress(f"WARNING: ALL DATA ON {self.device} WILL BE ERASED!")
self._report_progress(f"Unmounting disk {self.device} (force)...")
self._unmount_path(self.device, is_device=True, force=True)
time.sleep(2)
self._report_progress(f"Partitioning {self.device} with GPT scheme...")
self._run_command([
"diskutil", "partitionDisk", self.device, "GPT",
"MS-DOS FAT32", "EFI", "551MiB",
"JHFS+", "macOS_USB", "0b"
], timeout=180)
time.sleep(3)
esp_partition_dev = self._get_partition_device_id(self.device, "EFI")
macos_partition_dev = self._get_partition_device_id(self.device, "macOS_USB")
if not (esp_partition_dev and os.path.exists(esp_partition_dev)):
esp_partition_dev = f"{self.device}s1"
if not (macos_partition_dev and os.path.exists(macos_partition_dev)):
macos_partition_dev = f"{self.device}s2"
if not (os.path.exists(esp_partition_dev) and os.path.exists(macos_partition_dev)):
raise RuntimeError(f"Could not identify partitions on {self.device}. ESP: {esp_partition_dev}, macOS: {macos_partition_dev}")
self._report_progress(f"Identified ESP: {esp_partition_dev}, macOS Partition: {macos_partition_dev}")
# --- Write EFI content ---
self._report_progress(f"Converting OpenCore QCOW2 ({self.opencore_qcow2_path}) to RAW ({self.opencore_raw_path})...")
self._run_command(["qemu-img", "convert", "-O", "raw", self.opencore_qcow2_path, self.opencore_raw_path])
self._report_progress(f"Attaching RAW OpenCore image ({self.opencore_raw_path})...")
attach_cmd_efi = ["hdiutil", "attach", "-nomount", "-imagekey", "diskimage-class=CRawDiskImage", self.opencore_raw_path]
efi_attach_output = self._run_command(attach_cmd_efi, capture_output=True).stdout.strip()
raw_efi_disk_id = efi_attach_output.splitlines()[-1].strip().split()[0]
if not raw_efi_disk_id.startswith("/dev/disk"):
raise RuntimeError(f"Failed to attach raw EFI image: {efi_attach_output}")
self.attached_raw_images_devices.append(raw_efi_disk_id)
self._report_progress(f"Attached raw OpenCore image as {raw_efi_disk_id}")
time.sleep(2)
source_efi_partition_dev = self._get_partition_device_id(raw_efi_disk_id, "EFI") or f"{raw_efi_disk_id}s1"
self._report_progress(f"Mounting source EFI partition ({source_efi_partition_dev}) to {self.temp_opencore_mount}...")
self._run_command(["diskutil", "mount", "readOnly", "-mountPoint", self.temp_opencore_mount, source_efi_partition_dev], timeout=30)
self._report_progress(f"Mounting target USB ESP ({esp_partition_dev}) to {self.temp_usb_esp_mount}...")
self._run_command(["diskutil", "mount", "-mountPoint", self.temp_usb_esp_mount, esp_partition_dev], timeout=30)
source_efi_content_path = os.path.join(self.temp_opencore_mount, "EFI")
if not os.path.isdir(source_efi_content_path): source_efi_content_path = self.temp_opencore_mount
target_efi_dir_on_usb = os.path.join(self.temp_usb_esp_mount, "EFI")
self._report_progress(f"Copying EFI files from {source_efi_content_path} to {target_efi_dir_on_usb}...")
self._run_command(["sudo", "rsync", "-avh", "--delete", f"{source_efi_content_path}/", f"{target_efi_dir_on_usb}/"])
self._unmount_path(self.temp_opencore_mount, force=True)
self._unmount_path(self.temp_usb_esp_mount, force=True)
self._detach_raw_image_device(raw_efi_disk_id); raw_efi_disk_id = None
# --- Write macOS main image (File-level copy) ---
self._report_progress(f"Converting macOS QCOW2 ({self.macos_qcow2_path}) to RAW ({self.macos_raw_path})...")
self._report_progress("This may take a very long time...")
self._run_command(["qemu-img", "convert", "-O", "raw", self.macos_qcow2_path, self.macos_raw_path])
self._report_progress(f"Attaching RAW macOS image ({self.macos_raw_path})...")
attach_cmd_macos = ["hdiutil", "attach", "-nomount", "-imagekey", "diskimage-class=CRawDiskImage", self.macos_raw_path]
macos_attach_output = self._run_command(attach_cmd_macos, capture_output=True).stdout.strip()
raw_macos_disk_id = macos_attach_output.splitlines()[-1].strip().split()[0]
if not raw_macos_disk_id.startswith("/dev/disk"):
raise RuntimeError(f"Failed to attach raw macOS image: {macos_attach_output}")
self.attached_raw_images_devices.append(raw_macos_disk_id)
self._report_progress(f"Attached raw macOS image as {raw_macos_disk_id}")
time.sleep(2)
source_macos_part_dev = self._get_partition_device_id(raw_macos_disk_id, "Apple_APFS_Container") or \
self._get_partition_device_id(raw_macos_disk_id, "Apple_APFS") or \
self._get_partition_device_id(raw_macos_disk_id, "Apple_HFS") or \
f"{raw_macos_disk_id}s2"
if not (source_macos_part_dev and os.path.exists(source_macos_part_dev)):
raise RuntimeError(f"Could not find source macOS partition on {raw_macos_disk_id}")
self._report_progress(f"Mounting source macOS partition ({source_macos_part_dev}) to {self.temp_macos_source_mount}...")
self._run_command(["diskutil", "mount", "readOnly", "-mountPoint", self.temp_macos_source_mount, source_macos_part_dev], timeout=60)
self._report_progress(f"Mounting target USB macOS partition ({macos_partition_dev}) to {self.temp_usb_macos_target_mount}...")
self._run_command(["diskutil", "mount", "-mountPoint", self.temp_usb_macos_target_mount, macos_partition_dev], timeout=30)
self._report_progress(f"Copying macOS system files from {self.temp_macos_source_mount} to {self.temp_usb_macos_target_mount} (sudo rsync)...")
self._report_progress("This will also take a very long time.")
self._run_command([
"sudo", "rsync", "-avh", "--delete",
"--exclude=.Spotlight-V100", "--exclude=.fseventsd", "--exclude=/.Trashes", "--exclude=/System/Volumes/VM", "--exclude=/private/var/vm",
f"{self.temp_macos_source_mount}/", f"{self.temp_usb_macos_target_mount}/"
])
self._report_progress("USB writing process completed successfully.")
return True
except Exception as e:
self._report_progress(f"An error occurred during USB writing on macOS: {e}")
import traceback
self._report_progress(traceback.format_exc())
return False
finally:
self._cleanup_all_mounts_and_mappings()
self._cleanup_temp_files()
if __name__ == '__main__':
if platform.system() != "Darwin": print("This script is intended for macOS."); exit(1)
print("USB Writer macOS Standalone Test - File Copy Method")
mock_opencore_path = "mock_opencore_macos.qcow2"
mock_macos_path = "mock_macos_macos.qcow2"
if not os.path.exists(mock_opencore_path): subprocess.run(["qemu-img", "create", "-f", "qcow2", mock_opencore_path, "384M"])
if not os.path.exists(mock_macos_path): subprocess.run(["qemu-img", "create", "-f", "qcow2", mock_macos_path, "1G"])
print("\nAvailable disks (use 'diskutil list external physical' in Terminal to identify your USB):")
subprocess.run(["diskutil", "list", "external", "physical"], check=False)
test_device = input("\nEnter target disk identifier (e.g., /dev/diskX - NOT /dev/diskXsY). THIS DISK WILL BE WIPED: ")
if not test_device or not test_device.startswith("/dev/disk"):
print("Invalid disk identifier. Exiting.")
if os.path.exists(mock_opencore_path): os.remove(mock_opencore_path)
if os.path.exists(mock_macos_path): os.remove(mock_macos_path)
exit(1)
confirm = input(f"Are you sure you want to wipe {test_device} and write mock images? (yes/NO): ")
success = False
if confirm.lower() == 'yes':
print("Ensure you have sudo privileges for rsync if needed, or app is run as root.")
writer = USBWriterMacOS(test_device, mock_opencore_path, mock_macos_path, print)
success = writer.format_and_write()
else:
print("Test cancelled.")
print(f"Test finished. Success: {success}")
if os.path.exists(mock_opencore_path): os.remove(mock_opencore_path)
if os.path.exists(mock_macos_path): os.remove(mock_macos_path)
print("Mock files cleaned up.")