#!/usr/bin/env python3 import locale import curses import os import subprocess import re import time import zipfile from io import BytesIO import tempfile from pathlib import Path import json import threading, queue, contextlib from types import SimpleNamespace try: from functools import lru_cache except ImportError: lru_cache = lambda _: (lambda f: f) # fallback for Py<3.2 import atexit import shutil import datetime PROGRAM_NAME = "Stellar Commander" PROGRAM_VERSION = "0.4" NAVIGATION_HISTORY = {} CONFIG_PATH = Path.home() / '.stellarcommander.cfg' DRIVE_DEVICE = "8" CBM_FILE_TYPES = { 'PRG': 'Program', 'SEQ': 'Sequential', 'USR': 'User', 'REL': 'Relative', 'DEL': 'Deleted' } DEFAULT_CONFIG = { 'device_id': "8", 'connection_type': "serial1", 'connection_types': ["auto", "original", "serial1", "serial2", "parallel"] } FILE_COLORS = { 'DIR': 8, # Cyan 'PRG': 9, # Green 'D64': 10, # Magenta 'ZIP': 10, # Magenta (same as D64) 'SEQ': 11, # Yellow 'USR': 11, # Yellow 'DEL': 12, # Red 'REL': 8, # Cyan (treat like DIR) 'T64': 9, # Green } def load_config(): try: return json.loads(CONFIG_PATH.read_text()) except Exception: return DEFAULT_CONFIG.copy() def save_config(config): CONFIG_PATH.write_text(json.dumps(config)) def cleanup_temp_files(): tmp = Path(tempfile.gettempdir()) for path in tmp.glob("tmp*"): shutil.rmtree(path, ignore_errors=True) for path in tmp.glob("*.prg"): try: path.unlink() except Exception: pass def setup_colors(): curses.start_color() curses.use_default_colors() curses.init_pair(1, curses.COLOR_YELLOW, curses.COLOR_BLUE) # Header (yellow) curses.init_pair(2, curses.COLOR_BLACK, curses.COLOR_CYAN) # Footer curses.init_pair(3, curses.COLOR_BLACK, curses.COLOR_WHITE) # Highlight curses.init_pair(4, curses.COLOR_CYAN, -1) # Normal text curses.init_pair(5, curses.COLOR_YELLOW, curses.COLOR_BLUE) # Window border curses.init_pair(6, curses.COLOR_RED, -1) # Error text curses.init_pair(7, curses.COLOR_GREEN, -1) # Success text curses.init_pair(8, curses.COLOR_CYAN, -1) # Directories curses.init_pair(9, curses.COLOR_GREEN, -1) # PRG files curses.init_pair(10, curses.COLOR_MAGENTA, -1) # D64/ZIP archives curses.init_pair(11, curses.COLOR_YELLOW, -1) # SEQ/USR files curses.init_pair(12, curses.COLOR_RED, -1) # DEL files def list_dir(path): """Return list[dict] for path or ZIP file.""" try: if path.endswith('.zip'): with zipfile.ZipFile(path) as z: entries = [{'name': name, 'size': z.getinfo(name).file_size, 'type': 'ZIP', 'date': '-', 'is_dir': name.endswith('/')} for name in z.namelist() if not name.endswith('/')] entries.insert(0, {'name': '..', 'size': 0, 'type': 'DIR', 'date': '-', 'is_dir': True}) return entries # Normal directory listing entries = [] for name in os.listdir(path): full_path = os.path.join(path, name) is_dir = os.path.isdir(full_path) if is_dir: file_type = 'DIR' size = 0 else: ext = os.path.splitext(name)[1] file_type = ext[1:].upper() if ext else 'FILE' try: size = os.path.getsize(full_path) except OSError: size = 0 try: date = datetime.datetime.fromtimestamp( os.path.getmtime(full_path) ).strftime('%Y-%m-%d %H:%M') except OSError: date = '-' entries.append({ 'name': name, 'size': size, 'type': file_type, 'date': date, 'is_dir': is_dir }) # Sort: directories first, then by name (case-insensitive) entries.sort(key=lambda x: (not x['is_dir'], x['name'].lower())) entries.insert(0, { 'name': '..', 'size': 0, 'type': 'DIR', 'date': '-', 'is_dir': True }) return entries except Exception as e: return [{'name': '..', 'size': 0, 'type': 'DIR', 'date': '-', 'is_dir': True}] @lru_cache(maxsize=32) def extract_from_zip(zip_path, filename): try: with zipfile.ZipFile(zip_path) as z: with z.open(filename) as zf: content = zf.read() temp_dir = tempfile.mkdtemp() original_name = os.path.basename(filename) temp_path = os.path.join(temp_dir, original_name) with open(temp_path, 'wb') as f: f.write(content) return temp_path, original_name except Exception as e: print(f"Extraction error: {e}") return None, None def list_zip_contents(zip_path): try: with zipfile.ZipFile(zip_path) as z: return z.namelist() except: return [] def is_in_zip(current_path): return current_path.endswith('.zip') or '.zip/' in current_path def parse_cbm_directory(lines): parsed_lines = [] disk_title = "" disk_id = "" free_blocks = "" for line in lines: # Extract disk title and ID if line.startswith('0 ."'): match = re.search(r'^\d+\s+\.\"([^\"]*)\"\s+([^\s].*)$', line) if match: disk_title = match.group(1).strip() disk_id = match.group(2).strip() # Extract free blocks (keep this for footer) if 'blocks free' in line.lower(): free_blocks = line.strip() # Skip all header/footer/status lines (0, 1, 2 lines and blocks free) if not line.strip() or line.startswith(('0 ', '1 ', '2 ')) or ',' in line.lower() or 'blocks free' in line.lower(): continue # Parse and keep only file entries match = re.search(r'"([^"]+)"\s+([A-Z]{3})\s+(\d+)', line) if match: name, file_type, blocks = match.groups() file_type_desc = CBM_FILE_TYPES.get(file_type, file_type) parsed_lines.append(f'"{name}" {file_type} {blocks:>4} {file_type_desc}') else: parsed_lines.append(line) return parsed_lines, disk_title, disk_id, free_blocks def get_disk_directory(): try: result = subprocess.run( ["cbmctrl", "dir", DRIVE_DEVICE], capture_output=True, text=True, check=True ) lines = result.stdout.splitlines() return parse_cbm_directory(lines) except subprocess.CalledProcessError as e: return ["[Error reading CBM drive]"], "", "", "" def get_disk_status(): try: status_result = subprocess.run( ["cbmctrl", "status", DRIVE_DEVICE], capture_output=True, text=True, check=True ) return status_result.stdout.strip() except subprocess.CalledProcessError: return "[Status error]" def is_c64_program(filename): ext = os.path.splitext(filename)[1].lower() return ext in ('.prg', '.p00', '.t64') # Add support for P00 and T64 def ask_disk_name_and_id(stdscr): curses.echo() h, w = stdscr.getmaxyx() win_h, win_w = 9, 40 win_y = (h - win_h) // 2 win_x = (w - win_w) // 2 win = curses.newwin(win_h, win_w, win_y, win_x) win.border() win.addstr(1, 2, "🧽 C64 Disk Format", curses.A_BOLD) win.addstr(3, 2, "Name (max 16): ") win.refresh() name = win.getstr(3, 20, 16).decode("utf-8").strip() win.addstr(5, 2, "ID (e.g. 01): ") win.refresh() disk_id = win.getstr(5, 20, 2).decode("utf-8").strip() curses.noecho() return name, disk_id def transfer_d64_image(stdscr, path, source_is_drive, display_name=None, config=None): h, w = stdscr.getmaxyx() popup_h, popup_w = 10, 52 # Taller window for track display popup_y, popup_x = (h-popup_h)//2, (w-popup_w)//2 progress_win = curses.newwin(popup_h, popup_w, popup_y, popup_x) progress_win.border() if config is None: config = {'connection_type': 'serial1'} # Simple display logic - use exactly what we're given display = display_name if display_name else os.path.basename(path) direction = f"💾 {'Creating' if source_is_drive else 'Transferring'}: {display}" progress_win.addstr(1, 2, direction, curses.A_BOLD) # Set appropriate direction message if source_is_drive: direction = f"💾 Creating: {os.path.basename(path)}" cmd = ["d64copy", "-v", "-t", config['connection_type'], "-w", DRIVE_DEVICE, path] else: direction = f"💾 Transferring: {os.path.basename(path)}" cmd = ["d64copy", "-v", "-t", config['connection_type'], "-w", path, DRIVE_DEVICE] progress_win.addstr(1, 2, direction, curses.A_BOLD) filename = os.path.basename(path)[:popup_w-4] progress_win.addstr(1, 2, f"💾 Transferring: {display_name}", curses.A_BOLD) # Permanent progress elements progress_win.addstr(3, 2, "Overall: [") progress_win.addstr(3, popup_w-6, "] ") progress_win.addstr(5, 2, "Current Track: ") progress_win.addstr(7, 2, "Sectors: ") progress_win.refresh() try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, bufsize=1 ) total_sectors = 683 # Default for 35-track disks completed_sectors = 0 current_track = 0 while True: output = process.stdout.readline() if not output and process.poll() is not None: break # Parse track progress lines (e.g., " 3: *****-*****-*****-**- 8% 59/683") if re.match(r'^\s*\d+:', output): parts = output.split() try: # Update current track current_track = int(parts[0].strip(':')) # Update completed sectors if '/' in parts[-1]: completed_sectors = int(parts[-1].split('/')[0]) total_sectors = int(parts[-1].split('/')[1]) # Calculate percentage percent = min(100, (completed_sectors * 100) // total_sectors) # Update display progress_win.addstr(5, 17, f"{current_track}/35") progress_win.addstr(7, 10, f" {completed_sectors}/{total_sectors}") # Draw progress bar bar_width = popup_w - 17 filled = (percent * bar_width) // 100 progress_win.addstr(3, 12, "█" * filled + " " * (bar_width - filled)) progress_win.addstr(3, popup_w-8, f"] {percent}%") progress_win.refresh() except (ValueError, IndexError): pass # Skip malformed lines if process.returncode == 0: progress_win.addstr(8, 2, "✅ Transfer complete!", curses.color_pair(7)) else: raise subprocess.CalledProcessError(process.returncode, process.args) except Exception as e: progress_win.addstr(8, 2, f"❌ Error: {str(e)[:popup_w-10]}", curses.color_pair(6)) progress_win.refresh() progress_win.getch() return process.returncode == 0 if 'process' in locals() else False def ensure_min_terminal_size(stdscr): while True: h, w = stdscr.getmaxyx() if h >= 10 and w >= 40: return True try: stdscr.clear() msg = f"Need larger terminal (current: {w}x{h}, min: 40x10)" stdscr.addstr(0, 0, msg[:w-1]) stdscr.refresh() time.sleep(0.5) except: pass def show_config_menu(stdscr, config): h, w = stdscr.getmaxyx() win_h, win_w = 12, 50 win_y = (h - win_h) // 2 win_x = (w - win_w) // 2 win = curses.newwin(win_h, win_w, win_y, win_x) win.border() win.addstr(1, 2, "Configuration Menu", curses.A_BOLD) # Device ID selection win.addstr(3, 2, "1. C64 Device ID (8/9): ") win.addstr(3, 26, config['device_id'], curses.A_REVERSE) # Connection type selection win.addstr(5, 2, "2. Connection Type: ") # Initial display - no padding needed win.addstr(5, 22, config['connection_type'], curses.A_REVERSE) win.addstr(7, 2, "S. Save configuration") win.addstr(8, 2, "C. Cancel without saving") win.refresh() while True: key = win.getch() if key == ord('1'): config['device_id'] = '9' if config['device_id'] == '8' else '8' win.addstr(3, 26, config['device_id'], curses.A_REVERSE) elif key == ord('2'): # Clear the previous value win.addstr(5, 22, " " * 8) # Cycle through connection types if config['connection_type'] == "serial1": config['connection_type'] = "serial2" elif config['connection_type'] == "serial2": config['connection_type'] = "parallel" elif config['connection_type'] == "parallel": config['connection_type'] = "auto" elif config['connection_type'] == "auto": config['connection_type'] = "original" else: config['connection_type'] = "serial1" # Write new value without padding win.addstr(5, 22, config['connection_type'], curses.A_REVERSE) elif key in (ord('s'), ord('S')): save_config(config) return True # Config changed elif key in (ord('c'), ord('C')) or key == 27: # ESC return False # Config not changed win.refresh() def sizeof_fmt(num, suffix='B'): for unit in ['', 'K', 'M', 'G']: if abs(num) < 1024.0: return f"{num:3.1f}{unit}{suffix}" num /= 1024.0 return f"{num:.1f}T{suffix}" def draw_interface(stdscr, rel_row_left, rel_row_right, visible_files, current_path, visible_disk, focus, disk_title, disk_id, free_blocks, disk_status=None): try: stdscr.clear() h, w = stdscr.getmaxyx() # Ensure minimum terminal size if h < 10 or w < 40: stdscr.addstr(0, 0, "Terminal too small (min 40x10)") stdscr.refresh() return # Panel dimensions panel_start_y = 1 left_width = max(20, w // 2 - 1) right_start = min(w-2, left_width + 2) max_visible_rows = h - panel_start_y - 3 # Account for borders and footer # Column definitions col_name = max(25, int(left_width * 0.55)) # 55% for filename col_size = 8 col_type = 5 col_date = min(14, left_width - col_name - col_size - col_type - 3) # Header header_text = f" {PROGRAM_NAME} v{PROGRAM_VERSION}" header_x = max(1, (w - len(header_text)) // 2) stdscr.attron(curses.color_pair(1)) stdscr.addstr(0, 0, " " * w) stdscr.addstr(0, header_x, header_text[:w-1]) stdscr.attroff(curses.color_pair(1)) # Draw panel borders stdscr.attron(curses.color_pair(5)) # Left panel stdscr.addstr(panel_start_y, 0, "╔" + "═" * (left_width-2) + "╗") for y in range(panel_start_y + 1, h-2): stdscr.addch(y, 0, "║") stdscr.addch(y, left_width-1, "║") stdscr.addstr(h-2, 0, "╚" + "═" * (left_width-2) + "╝") # Right panel stdscr.addstr(panel_start_y, right_start, "╔" + "═" * (w-right_start-2) + "╗") for y in range(panel_start_y + 1, h-2): stdscr.addch(y, right_start, "║") stdscr.addch(y, w-1, "║") stdscr.addstr(h-2, right_start, "╚" + "═" * (w-right_start-2) + "╝") stdscr.attroff(curses.color_pair(5)) # Panel headers stdscr.attron(curses.color_pair(1)) # Left header pc_header = f"PC: {current_path[:left_width-10]}" stdscr.addstr(panel_start_y, 2, pc_header[:left_width-2]) # Right header c64_header = f"C64: Drive {DRIVE_DEVICE}" if disk_title or disk_id: c64_header += " [" if disk_title: c64_header += f"'{disk_title}'" if disk_id: c64_header += f" {disk_id}" c64_header += "]" stdscr.addstr(panel_start_y, right_start+2, c64_header[:w-right_start-4]) stdscr.attroff(curses.color_pair(1)) # Left panel files - CRITICAL FIXES HERE if focus == "left": stdscr.attron(curses.color_pair(1)) header = ("Name".ljust(col_name)[:col_name] + "Size".ljust(col_size)[:col_size] + " " + "Type".ljust(col_type)[:col_type] + " " + "Modified".ljust(col_date)[:col_date]) stdscr.addstr(panel_start_y+1, 2, header[:left_width-2]) if focus == "left": stdscr.attroff(curses.color_pair(1)) for display_idx, entry in enumerate(visible_files): y = panel_start_y + 2 + display_idx if y >= h - 2: # do not draw past bottom border break # highlight uses the SAME y # Highlight current row - FIXED IMPLEMENTATION if focus == "left" and display_idx == rel_row_left: stdscr.attron(curses.color_pair(3)) # Clear entire line first stdscr.addstr(y, 1, " " * (left_width-2), curses.color_pair(3)) # Now draw the content name = (entry['name'] + "/" if entry['is_dir'] else entry['name'])[:col_name-1] size = sizeof_fmt(entry['size'])[:col_size-1] if not entry['is_dir'] else "-".center(col_size) ftype = entry['type'][:col_type-1] date = entry['date'][:col_date-1] line = (name.ljust(col_name) + size.rjust(col_size) + " " + ftype.ljust(col_type) + " " + date.ljust(col_date)) stdscr.addstr(y, 2, line[:left_width-2]) stdscr.attroff(curses.color_pair(3)) else: # Normal drawing for non-highlighted rows color_pair = FILE_COLORS.get(entry['type'], 4) name = (entry['name'] + "/" if entry['is_dir'] else entry['name'])[:col_name-1] size = sizeof_fmt(entry['size'])[:col_size-1] if not entry['is_dir'] else "-".center(col_size) ftype = entry['type'][:col_type-1] date = entry['date'][:col_date-1] line = (name.ljust(col_name) + size.rjust(col_size) + " " + ftype.ljust(col_type) + " " + date.ljust(col_date)) stdscr.attron(curses.color_pair(color_pair)) stdscr.addstr(y, 2, line[:left_width-2]) stdscr.attroff(curses.color_pair(color_pair)) # Right panel disk contents - SIMILAR FIXES for display_idx, line in enumerate(visible_disk): y = panel_start_y + 1 + display_idx if y >= h - 3: # Don't draw beyond window bottom break try: if focus == "right" and display_idx == rel_row_right: stdscr.attron(curses.color_pair(3)) # Clear entire line first stdscr.addstr(y, right_start+1, " " * (w-right_start-4), curses.color_pair(3)) # Now draw the content stdscr.addstr(y, right_start+2, line[:w-right_start-4]) stdscr.attroff(curses.color_pair(3)) else: stdscr.attron(curses.color_pair(4)) stdscr.addstr(y, right_start+2, line[:w-right_start-4]) stdscr.attroff(curses.color_pair(4)) except curses.error: continue # Status bar if h >= 3 and w >= 10: status_display = "" if free_blocks: status_display = free_blocks[:w//2] if disk_status and disk_status != "[Status error]": status_display = (status_display + " | " + disk_status)[:w-right_start-4] stdscr.attron(curses.color_pair(4)) stdscr.addstr(h-3, right_start+2, status_display) stdscr.attroff(curses.color_pair(4)) # Footer if h >= 1 and w >= 10: footer_text = ("F1 Help | F2 Config | F3 Refresh | F4 Image | " "F5 Copy | F6 Format | F8 Delete | TAB Switch | Q Quit") stdscr.attron(curses.color_pair(2)) stdscr.addstr(h-1, 0, footer_text[:w-1]) stdscr.attroff(curses.color_pair(2)) stdscr.refresh() except Exception as e: try: stdscr.clear() stdscr.addstr(0, 0, f"Display error: {str(e)[:30]}") stdscr.refresh() except: pass def show_progress_popup(stdscr, message): h, w = stdscr.getmaxyx() lines = message.splitlines() win_h = len(lines) + 4 win_w = max(len(line) for line in lines) + 6 win_y = (h - win_h) // 2 win_x = (w - win_w) // 2 win = curses.newwin(win_h, win_w, win_y, win_x) win.border() for i, line in enumerate(lines): win.addstr(2 + i, 3, line) win.refresh() def show_help_popup(stdscr): h, w = stdscr.getmaxyx() win_h, win_w = 16, 50 win_y = (h - win_h) // 2 win_x = (w - win_w) // 2 help_win = curses.newwin(win_h, win_w, win_y, win_x) help_win.border() help_win.addstr(1, 2, "Stellar Commander Help", curses.A_BOLD) help_win.addstr(3, 2, "TAB - Switch between PC and C64 drive") help_win.addstr(4, 2, "F1 - Show this help") help_win.addstr(5, 2, "F2 - Configuration") help_win.addstr(6, 2, "F3 - Refresh directory listing") help_win.addstr(7, 2, "F4 - Write .d64 image (either way)") help_win.addstr(8, 2, "F5 - Copy selected file (either way)") help_win.addstr(9, 2, "F6 - Format C64 disk") help_win.addstr(10, 2, "F8 - Delete selected file") help_win.addstr(11, 2, "Arrow keys - Navigate files") help_win.addstr(12, 2, "Enter - Open directory") help_win.addstr(14, 2, "Press any key to close") help_win.refresh() help_win.getch() def show_info_popup(stdscr, message): lines = message.strip().splitlines() h, w = stdscr.getmaxyx() max_line_length = max(len(line.encode("utf-8")) for line in lines) win_h = len(lines) + 4 win_w = max_line_length + 6 win_y = (h - win_h) // 2 win_x = (w - win_w) // 2 win = curses.newwin(win_h, win_w, win_y, win_x) win.box() # Color error/success messages appropriately if "❌" in message: color = curses.color_pair(6) elif "✅" in message: color = curses.color_pair(7) else: color = curses.color_pair(4) for i, line in enumerate(lines): win.addstr(2 + i, 3, line, color) win.addstr(win_h - 2, (win_w - 16) // 2, "[Press any key]") win.refresh() win.getch() def cleanup_navigation_history(current_path): global NAVIGATION_HISTORY # Keep only paths that are parents of the current path NAVIGATION_HISTORY = {path: pos for path, pos in NAVIGATION_HISTORY.items() if current_path.startswith(path)} def handle_f1_key(stdscr): show_help_popup(stdscr) def handle_f2_key(stdscr, config): config_changed = show_config_menu(stdscr, config) if config_changed: DRIVE_DEVICE = config['device_id'] disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() return config_changed, disk_dir, disk_title, disk_id, free_blocks def handle_f3_key(stdscr, current_path): disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() files = list_dir(current_path) return disk_dir, disk_title, disk_id, free_blocks, files def handle_f4_key(stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, config, free_blocks): # Initialize all return variables disk_title = "" disk_id = "" free_blocks = "" if focus == "left": selected_entry = files[current_row_left] selected_name = selected_entry['name'] full_path = os.path.join(current_path, selected_name) if current_path.endswith('.zip') and selected_name.lower().endswith('.d64'): try: temp_path, original_name = extract_from_zip(current_path, selected_name) if not temp_path: show_info_popup(stdscr, "❌ Failed to extract from ZIP!") return h, w = stdscr.getmaxyx() win = curses.newwin(7, 50, (h-7)//2, (w-50)//2) win.border() win.addstr(1, 2, "Transfer from ZIP archive", curses.A_BOLD) win.addstr(2, 2, f"ZIP: {os.path.basename(current_path)}") win.addstr(3, 2, f"Image: {selected_name[:40]}") win.addstr(5, 2, "Confirm? (Y/N)") win.refresh() confirm = win.getch() in (ord('y'), ord('Y')) if confirm: if transfer_d64_image(stdscr, temp_path, source_is_drive=False, display_name=original_name, config=config): disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() if temp_path and os.path.exists(temp_path): os.unlink(temp_path) temp_dir = os.path.dirname(temp_path) if os.path.exists(temp_dir): os.rmdir(temp_dir) except Exception as e: show_info_popup(stdscr, f"❌ ZIP extraction failed:\n{str(e)}") elif selected_name.lower().endswith('.d64'): if not os.path.isfile(full_path): show_info_popup(stdscr, "❌ File not found!") else: h, w = stdscr.getmaxyx() win = curses.newwin(6, 50, (h-6)//2, (w-50)//2) win.border() win.addstr(1, 2, "Transfer .d64 image to C64 disk?", curses.A_BOLD) win.addstr(3, 2, f"File: {selected_name[:40]}") win.addstr(4, 2, "Confirm? (Y/N)") win.refresh() confirm = win.getch() if confirm in (ord('y'), ord('Y')): if transfer_d64_image(stdscr, full_path, source_is_drive=False, display_name=selected_name[:40], config=config): disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() else: show_info_popup(stdscr, "❌ Not a .d64 disk image!\nPlease select a .d64 file") elif focus == "right" and is_in_zip(current_path): show_info_popup(stdscr, "❌ Function disabled\n\nCannot transfer disk images\nwhile browsing inside ZIP archive") elif focus == "right": h, w = stdscr.getmaxyx() win = curses.newwin(7, 52, (h-7)//2, (w-52)//2) win.border() win.addstr(1, 2, "💾 Create Disk Image", curses.A_BOLD) win.addstr(3, 2, "Save as (.d64): ") curses.echo() filename = "" try: filename = win.getstr(3, 18, 16).decode('utf-8').strip() if not filename.lower().endswith('.d64'): filename += '.d64' target_path = os.path.join(current_path, filename) if os.path.exists(target_path): show_info_popup(stdscr, f"❌ File exists!\n{filename}") else: if transfer_d64_image(stdscr, target_path, source_is_drive=True, display_name=filename, config=config): files = list_dir(current_path) finally: curses.noecho() return disk_dir, disk_title, disk_id, free_blocks, files def handle_f5_key(stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, config, free_blocks): # Initialize return variables disk_title = "" disk_id = "" temp_path = None def update_progress_popup(stdscr, message, progress, total): height, width = stdscr.getmaxyx() popup_h = 7 # Reduced height since we removed the separator popup_w = min(64, width - 4) # Slightly wider to accommodate elements y = (height - popup_h) // 2 x = (width - popup_w) // 2 # Calculate percentage and block progress percent = min(100, int((progress / total) * 100)) if total > 0 else 0 bar_width = popup_w - 12 # Adjusted to ensure everything fits filled = int(round(bar_width * progress / total)) if total > 0 else 0 # Create popup window win = curses.newwin(popup_h, popup_w, y, x) win.border() # Add message (split into two lines if needed) msg_lines = message.split('\n') for i, line in enumerate(msg_lines[:2]): # Show max 2 lines win.addstr(1 + i, 2, line[:popup_w-4]) # Add progress bar with proper spacing progress_bar = f"[{'█' * filled}{' ' * (bar_width - filled)}] {percent}%" win.addstr(3, 2, progress_bar) # Add transfer stats below if total > 0: win.addstr(4, 2, f"Transferred: {progress}/{total} blocks") win.refresh() if focus == "left": selected_entry = files[current_row_left] selected_name = selected_entry['name'] full_path = os.path.join(current_path, selected_name) if current_path.endswith('.zip'): if not is_c64_program(selected_name): show_info_popup(stdscr, "❌ Not a C64 program!\nOnly .PRG/.T64/.P00 files can be copied") return disk_dir, disk_title, disk_id, free_blocks, files try: temp_path, original_name = extract_from_zip(current_path, selected_name) if not temp_path: show_info_popup(stdscr, "❌ Failed to extract from ZIP!") return disk_dir, disk_title, disk_id, free_blocks, files # Get file size for progress tracking file_size = os.path.getsize(temp_path) # Different block calculations for different file types if selected_name.lower().endswith('.t64'): # T64 files have container overhead, estimate actual program size # T64 header is typically 64 bytes + directory entries # Subtract overhead and calculate blocks estimated_program_size = max(0, file_size - 64) # Subtract T64 header blocks = (estimated_program_size + 253) // 254 elif selected_name.lower().endswith('.p00'): # P00 files have a 26-byte header estimated_program_size = max(0, file_size - 26) # Subtract P00 header blocks = (estimated_program_size + 253) // 254 else: # PRG files: direct calculation blocks = (file_size + 253) // 254 # C64 blocks (254 bytes data each) show_progress_popup(stdscr, f"⏳ Copying to C64\n{selected_name}...") # Start subprocess with stdout/stderr capture process = subprocess.Popen( ["cbmwrite", "-v", "-t", config['connection_type'], DRIVE_DEVICE, temp_path, selected_name], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, # Line buffered universal_newlines=True ) # Parse output for progress transferred_blocks = 0 for line in process.stdout: if "send byte count" in line: transferred_blocks += 1 update_progress_popup(stdscr, f"⏳ Copying to C64\n{selected_name}", transferred_blocks, blocks) # Wait for process to complete return_code = process.wait() if return_code == 0: show_info_popup(stdscr, f"✅ Program copied to C64:\n{selected_name}\n\n{blocks} blocks transferred") disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() else: raise subprocess.CalledProcessError(return_code, process.args) except subprocess.CalledProcessError as e: err = e.stderr.strip() if e.stderr else e.stdout.strip() if e.stdout else "Unknown error" show_info_popup(stdscr, f"❌ Copy error:\n{err}") finally: if temp_path and os.path.exists(temp_path): os.unlink(temp_path) else: if os.path.isdir(full_path): show_info_popup(stdscr, "❌ Cannot copy directories!") elif not is_c64_program(selected_name): show_info_popup(stdscr, "❌ Not a C64 program!\nOnly .PRG/.T64/.P00 files can be copied") else: # Get file size for progress tracking file_size = os.path.getsize(full_path) # Different block calculations for different file types if selected_name.lower().endswith('.t64'): # T64 files have container overhead, estimate actual program size # T64 header is typically 64 bytes + directory entries # Subtract overhead and calculate blocks estimated_program_size = max(0, file_size - 64) # Subtract T64 header blocks = (estimated_program_size + 253) // 254 elif selected_name.lower().endswith('.p00'): # P00 files have a 26-byte header estimated_program_size = max(0, file_size - 26) # Subtract P00 header blocks = (estimated_program_size + 253) // 254 else: # PRG files: direct calculation blocks = (file_size + 253) // 254 # C64 blocks (254 bytes data each) show_progress_popup(stdscr, f"⏳ Copying to C64\n{selected_name}...") try: # Start subprocess with stdout/stderr capture process = subprocess.Popen( ["cbmwrite", "-v", "-t", config['connection_type'], DRIVE_DEVICE, full_path, selected_name], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, # Line buffered universal_newlines=True ) # Parse output for progress transferred_blocks = 0 for line in process.stdout: if "send byte count" in line: transferred_blocks += 1 update_progress_popup(stdscr, f"⏳ Copying to C64\n{selected_name}", transferred_blocks, blocks) # Wait for process to complete return_code = process.wait() if return_code == 0: show_info_popup(stdscr, f"✅ Program copied to C64:\n{selected_name}\n\n{blocks} blocks transferred") disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() else: raise subprocess.CalledProcessError(return_code, process.args) except subprocess.CalledProcessError as e: err = e.stderr.strip() if e.stderr else e.stdout.strip() if e.stdout else "Unknown error" show_info_popup(stdscr, f"❌ Copy error:\n{err}") elif focus == "right" and is_in_zip(current_path): show_info_popup(stdscr, "❌ Function disabled\n\nCannot copy files from C64\nwhile browsing inside ZIP archive") elif focus == "right": if current_row_right < len(disk_dir): selected_line = disk_dir[current_row_right] # Improved parsing that handles C64 directory format with spaces parts = selected_line.split('"') if len(parts) >= 3: quoted_name = parts[1] # The filename between quotes remaining_parts = parts[2].split() if len(remaining_parts) >= 1: filetype = remaining_parts[0].upper() # The filetype (PRG, SEQ, etc.) else: filetype = 'PRG' # Default if not specified # Create clean PC filename clean_name = " ".join(quoted_name.strip().split()) # Normalize spaces pc_filename = f"{clean_name}.{filetype.lower() if filetype != 'PRG' else 'prg'}" target_path = os.path.join(current_path, pc_filename) if not os.path.dirname(target_path) == os.path.abspath(current_path): show_info_popup(stdscr, f"❌ Invalid target path!\n{target_path}") return disk_dir, disk_title, disk_id, free_blocks, files if os.path.exists(target_path): show_info_popup(stdscr, f"❌ File exists:\n{pc_filename}\n\nOverwrite not implemented") return disk_dir, disk_title, disk_id, free_blocks, files # FIXED: Better parsing of C64 directory format to extract block count # C64 directory format: " 123 "FILENAME " PRG" # We need to extract the number at the beginning of the line blocks = 0 try: # Split the original line by spaces and look for the first numeric part line_parts = selected_line.strip().split() for part in line_parts: if part.isdigit(): blocks = int(part) break # If we couldn't find blocks in the line, try to estimate from line position if blocks == 0: # Alternative: look for number before the first quote before_quote = selected_line.split('"')[0].strip() if before_quote.isdigit(): blocks = int(before_quote) else: # Last resort: extract any digits from the beginning import re match = re.search(r'^\s*(\d+)', selected_line) if match: blocks = int(match.group(1)) except (ValueError, IndexError): blocks = 0 # Fallback to 0 if parsing fails show_progress_popup(stdscr, f"⏳ Copying from C64\n{pc_filename}...") try: # Start subprocess with stderr capture (cbmread outputs debug info to stderr) process = subprocess.Popen( ["cbmread", "-v", "-t", config['connection_type'], DRIVE_DEVICE, quoted_name, target_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, # Line buffered universal_newlines=True, encoding='ascii', errors='ignore' ) # Read stderr in a non-blocking way using threading import threading import queue stderr_queue = queue.Queue() def read_stderr(): try: for line in process.stderr: stderr_queue.put(line) except: pass stderr_queue.put(None) # Signal end # Start stderr reading thread stderr_thread = threading.Thread(target=read_stderr) stderr_thread.daemon = True stderr_thread.start() # Parse stderr output for progress - Look for the right pattern transferred_blocks = 0 debug_lines = [] while True: try: line = stderr_queue.get(timeout=0.1) if line is None: # End signal break debug_lines.append(line.strip()) # Look for the specific pattern from the log: "number of bytes read for block X:" if "number of bytes read for block" in line: transferred_blocks += 1 update_progress_popup(stdscr, f"⏳ Copying from C64\n{pc_filename}", transferred_blocks, blocks) except queue.Empty: # Check if process is still running if process.poll() is not None: break continue # Wait for process to complete return_code = process.wait() if return_code == 0: show_info_popup(stdscr, f"✅ File copied from C64:\n{pc_filename}\n\n{transferred_blocks} blocks transferred") files = list_dir(current_path) else: raise subprocess.CalledProcessError(return_code, process.args) except subprocess.CalledProcessError as e: err = e.stderr.strip() if e.stderr else e.stdout.strip() if e.stdout else "Unknown error" show_info_popup(stdscr, f"❌ Copy error:\n{err}") else: show_info_popup(stdscr, f"❌ Could not parse:\n{selected_line}") return disk_dir, disk_title, disk_id, free_blocks, files def handle_f6_key(stdscr, refresh_disk_dir): name, disk_id = ask_disk_name_and_id(stdscr) if not name or not disk_id: show_info_popup(stdscr, "❌ Disk name and ID are required!") return # Progress window setup height, width = stdscr.getmaxyx() win_height = 8 win_width = 60 win_y = (height - win_height) // 2 win_x = (width - win_width) // 2 progress_win = curses.newwin(win_height, win_width, win_y, win_x) progress_win.box() progress_win.addstr(1, 2, f"💾 Formatting: {name},{disk_id}") progress_win.addstr(2, 2, f"Drive: {DRIVE_DEVICE}") progress_win.addstr(3, 2, "Progress:") progress_win.addstr(4, 2, "[" + " " * 50 + "]") progress_win.addstr(5, 2, "Status: Initializing...") progress_win.refresh() try: # Reset drive first subprocess.run(["cbmctrl", "reset"], check=True) # Update status progress_win.addstr(5, 2, "Status: Starting format... ") progress_win.refresh() # Start format process with progress option process = subprocess.Popen( ["cbmformat", "-p", "-s", DRIVE_DEVICE, f"{name},{disk_id}"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) progress_chars = 0 final_status = "" # Read output line by line while True: output = process.stdout.read(1) if not output: break if output == '#': progress_chars += 1 # Update progress bar (assuming max 35 # characters based on your example) bar_length = min(50, int((progress_chars / 35) * 50)) progress_bar = "█" * bar_length + " " * (50 - bar_length) progress_win.addstr(4, 3, progress_bar) progress_win.addstr(5, 2, f"Status: Formatting... {progress_chars}/35") progress_win.refresh() elif output == '\n': # Read the status line after the progress remaining = process.stdout.readline() if remaining and any(s in remaining for s in ["00,", "20,", "21,", "74,"]): final_status = remaining.strip() break # Wait for process to complete process.wait() if process.returncode == 0: # Complete progress bar progress_win.addstr(4, 3, "█" * 50) progress_win.addstr(5, 2, f"Status: {final_status or '00, OK'} ") progress_win.addstr(6, 2, "Press any key to continue...") progress_win.refresh() stdscr.getch() # Wait for keypress else: stderr_output = process.stderr.read() err = stderr_output.strip() or "Unknown error" show_info_popup(stdscr, f"❌ Format error\n\n{err}") except Exception as e: show_info_popup(stdscr, f"❌ Format error\n\n{str(e)}") finally: # Clean up progress window del progress_win disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() stdscr.refresh() return disk_dir, disk_title, disk_id, free_blocks def handle_f8_key(stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, free_blocks): # Initialize variables we'll return disk_title = "" disk_id = "" if focus == "left": if 0 <= current_row_left < len(files): selected_entry = files[current_row_left] filename = selected_entry['name'] full_path = os.path.join(current_path, filename) if os.path.isfile(full_path): try: os.remove(full_path) show_info_popup(stdscr, f"🗑️ File deleted:\n{filename}") files = list_dir(current_path) except Exception as e: show_info_popup(stdscr, f"❌ Delete error:\n{e}") else: show_info_popup(stdscr, "❌ Not a regular file!") elif focus == "right": if 0 <= current_row_right < len(disk_dir): selected_line = disk_dir[current_row_right] match = re.search(r'"([^"]+)"', selected_line) if match: name_only = match.group(1) try: subprocess.run( ["cbmctrl", "pcommand", DRIVE_DEVICE, f's0:{name_only}'], check=True ) show_progress_popup(stdscr, f"Deleting file on C64:\n{name_only}...") status = subprocess.run( ["cbmctrl", "status", DRIVE_DEVICE], capture_output=True, text=True, check=True ) status_line = status.stdout.strip() if "files scratched" in status_line.lower(): show_info_popup(stdscr, f"🗑️ C64 file deleted:\n{name_only}\n\nStatus:\n{status_line}") # Refresh directory after successful delete disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() else: show_info_popup(stdscr, f"❌ Error:\n{status_line}") except subprocess.CalledProcessError as e: err = e.stderr.strip() or e.stdout.strip() or str(e) show_info_popup(stdscr, f"❌ Delete error:\n{err}") else: show_info_popup(stdscr, "❌ Could not parse filename!") return disk_dir, disk_title, disk_id, free_blocks, files def main(stdscr): setup_colors() curses.curs_set(0) current_path = os.getcwd() files = list_dir(current_path) current_row_left = 0 current_row_right = 0 focus = "left" top_row_left = 0 top_row_right = 0 config = load_config() global DRIVE_DEVICE DRIVE_DEVICE = config['device_id'] disk_dir, disk_title, disk_id, free_blocks = get_disk_directory() while True: h, w = stdscr.getmaxyx() max_visible_rows = h - 5 # Account for header/footer/borders # Ensure current rows are within bounds current_row_left = max(0, min(current_row_left, len(files) - 1)) current_row_right = max(0, min(current_row_right, len(disk_dir) - 1)) # CRITICAL FIX: Adjust top rows to keep cursor visible when scrolling down if focus == "left": if current_row_left < top_row_left: top_row_left = current_row_left elif current_row_left >= top_row_left + max_visible_rows: top_row_left = max(0, current_row_left - max_visible_rows + 1) # Ensure we don't show empty space at bottom top_row_left = min(top_row_left, max(0, len(files) - max_visible_rows)) else: # focus == "right" if current_row_right < top_row_right: top_row_right = current_row_right elif current_row_right >= top_row_right + max_visible_rows: top_row_right = max(0, current_row_right - max_visible_rows + 1) top_row_right = min(top_row_right, max(0, len(disk_dir) - max_visible_rows)) # Get visible portions visible_files = files[top_row_left:top_row_left + max_visible_rows] visible_disk = disk_dir[top_row_right:top_row_right + max_visible_rows] rel_row_left = 0 rel_row_right = 0 # Calculate relative cursor positions if focus == "left": rel_row_left = current_row_left - top_row_left # Ensure cursor stays within visible area when at bottom if rel_row_left >= max_visible_rows: rel_row_left = max_visible_rows - 1 top_row_left = current_row_left - rel_row_left else: rel_row_right = current_row_right - top_row_right if rel_row_right >= max_visible_rows: rel_row_right = max_visible_rows - 1 top_row_right = current_row_right - rel_row_right draw_interface(stdscr, rel_row_left, rel_row_right, visible_files, current_path, visible_disk, focus, disk_title, disk_id, free_blocks, get_disk_status()) key = stdscr.getch() if key == 9: # TAB focus = "right" if focus == "left" else "left" if focus == "left": if key == curses.KEY_UP: current_row_left = max(0, current_row_left - 1) # Keep cursor in view if current_row_left < top_row_left: top_row_left = current_row_left elif key == curses.KEY_DOWN: if current_row_left < len(files) - 1: current_row_left += 1 # Keep cursor in view if current_row_left >= top_row_left + max_visible_rows: top_row_left += 1 elif key == curses.KEY_PPAGE: # PAGE UP current_row_left = max(0, current_row_left - max_visible_rows) top_row_left = max(0, top_row_left - max_visible_rows) # Ensure cursor is visible if current_row_left < top_row_left: top_row_left = current_row_left elif key == curses.KEY_NPAGE: # PAGE DOWN current_row_left = min(len(files) - 1, current_row_left + max_visible_rows) # Adjust viewport if needed if current_row_left >= top_row_left + max_visible_rows: top_row_left = current_row_left - max_visible_rows + 1 # Don't scroll past end top_row_left = min(top_row_left, len(files) - max_visible_rows) elif key == curses.KEY_HOME: current_row_left = 0 top_row_left = 0 elif key == curses.KEY_END: current_row_left = len(files) - 1 top_row_left = max(0, len(files) - max_visible_rows) elif key in [curses.KEY_ENTER, 10, 13]: selected_entry = files[current_row_left] selected_name = selected_entry['name'] NAVIGATION_HISTORY[current_path] = current_row_left if selected_name == "..": parent_path = os.path.dirname(current_path) current_path = parent_path files = list_dir(current_path) current_row_left = NAVIGATION_HISTORY.get(current_path, 0) top_row_left = 0 elif selected_name.endswith('.zip') or (current_path.endswith('.zip') and not selected_entry['is_dir']): current_path = os.path.join(current_path, selected_name) files = list_dir(current_path) current_row_left = 0 top_row_left = 0 elif selected_entry['is_dir']: current_path = os.path.join(current_path, selected_name) files = list_dir(current_path) current_row_left = 0 top_row_left = 0 cleanup_navigation_history(current_path) else: # focus == "right" if key == curses.KEY_UP: current_row_right = max(0, current_row_right - 1) if current_row_right < top_row_right: top_row_right = current_row_right elif key == curses.KEY_DOWN: if current_row_right < len(disk_dir) - 1: current_row_right += 1 if current_row_right >= top_row_right + max_visible_rows: top_row_right += 1 elif key == curses.KEY_PPAGE: # PAGE UP current_row_right = max(0, current_row_right - max_visible_rows) top_row_right = max(0, top_row_right - max_visible_rows) if current_row_right < top_row_right: top_row_right = current_row_right elif key == curses.KEY_NPAGE: # PAGE DOWN current_row_right = min(len(disk_dir) - 1, current_row_right + max_visible_rows) if current_row_right >= top_row_right + max_visible_rows: top_row_right = current_row_right - max_visible_rows + 1 top_row_right = min(top_row_right, len(disk_dir) - max_visible_rows) elif key == curses.KEY_HOME: current_row_right = 0 top_row_right = 0 elif key == curses.KEY_END: current_row_right = len(disk_dir) - 1 top_row_right = max(0, len(disk_dir) - max_visible_rows) if key == curses.KEY_F1: handle_f1_key(stdscr) elif key == curses.KEY_F2: config_changed, disk_dir, disk_title, disk_id, free_blocks = handle_f2_key(stdscr, config) elif key == curses.KEY_F3: disk_dir, disk_title, disk_id, free_blocks, files = handle_f3_key(stdscr, current_path) top_row_left = 0 elif key == curses.KEY_F4: disk_dir, disk_title, disk_id, free_blocks, files = handle_f4_key( stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, config, free_blocks) elif key == curses.KEY_F5: disk_dir, disk_title, disk_id, free_blocks, files = handle_f5_key( stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, config, free_blocks) elif key == curses.KEY_F6: disk_dir, disk_title, disk_id, free_blocks = handle_f6_key(stdscr, lambda: get_disk_directory()) elif key == curses.KEY_F8: disk_dir, disk_title, disk_id, free_blocks, files = handle_f8_key( stdscr, focus, files, current_row_left, current_path, current_row_right, disk_dir, free_blocks) elif key == ord('q'): break if __name__ == "__main__": curses.wrapper(main) atexit.register(cleanup_temp_files)