From 0097b245a92b9e4a4a09e1155ae22ca7ba60c3c6 Mon Sep 17 00:00:00 2001 From: Anshgrover23 Date: Thu, 15 Jan 2026 17:20:57 +0530 Subject: [PATCH 1/2] Refactor: type hints and imports across multiple modules --- cortex/benchmark.py | 88 ++++++++++--------- cortex/branding.py | 24 +++--- cortex/cli.py | 54 +++++++----- cortex/gpu_manager.py | 84 +++++++++--------- cortex/health_score.py | 26 ++---- cortex/licensing.py | 53 +++++++----- cortex/output_formatter.py | 49 +++++------ cortex/printer_setup.py | 159 ++++++++++++++++++++--------------- cortex/semver_resolver.py | 37 +++----- cortex/stdin_handler.py | 30 +++---- cortex/systemd_helper.py | 124 ++++++++++++++------------- cortex/update_checker.py | 31 ++++--- cortex/updater.py | 19 +++-- cortex/version_manager.py | 6 +- cortex/wifi_driver.py | 30 ++++--- tests/test_benchmark.py | 39 ++------- tests/test_gpu_manager.py | 31 +++---- tests/test_health_score.py | 2 +- tests/test_printer_setup.py | 12 ++- tests/test_stdin_handler.py | 24 ++---- tests/test_systemd_helper.py | 21 ++--- tests/test_update_checker.py | 6 +- tests/test_updater.py | 2 +- tests/test_wifi_driver.py | 6 +- 24 files changed, 479 insertions(+), 478 deletions(-) diff --git a/cortex/benchmark.py b/cortex/benchmark.py index 1b35b45b3..8687db0fc 100644 --- a/cortex/benchmark.py +++ b/cortex/benchmark.py @@ -15,15 +15,15 @@ from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path -from typing import Any, List, Optional, Tuple +from typing import Any, Optional from rich import box from rich.console import Console from rich.panel import Panel -from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn +from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn from rich.table import Table -from cortex.branding import console, cx_print, cx_header, CORTEX_CYAN +from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print # Model recommendations based on system capabilities MODEL_REQUIREMENTS = { @@ -57,11 +57,11 @@ class BenchmarkReport: timestamp: str = "" system_info: dict = field(default_factory=dict) - results: List[BenchmarkResult] = field(default_factory=list) + results: list[BenchmarkResult] = field(default_factory=list) overall_score: int = 0 rating: str = "" - can_run: List[str] = field(default_factory=list) - needs_upgrade: List[str] = field(default_factory=list) + can_run: list[str] = field(default_factory=list) + needs_upgrade: list[str] = field(default_factory=list) upgrade_suggestion: str = "" def to_dict(self) -> dict[str, Any]: @@ -95,7 +95,7 @@ class CortexBenchmark: def __init__(self, verbose: bool = False): self.verbose = verbose - self._results: List[BenchmarkResult] = [] + self._results: list[BenchmarkResult] = [] def _get_system_info(self) -> dict: """Gather basic system information.""" @@ -118,7 +118,9 @@ def _get_system_info(self) -> dict: elif platform.system() == "Darwin": result = subprocess.run( ["sysctl", "-n", "machdep.cpu.brand_string"], - capture_output=True, text=True, timeout=5 + capture_output=True, + text=True, + timeout=5, ) if result.returncode == 0: info["cpu_model"] = result.stdout.strip() @@ -139,8 +141,7 @@ def _get_system_info(self) -> dict: break elif platform.system() == "Darwin": result = subprocess.run( - ["sysctl", "-n", "hw.memsize"], - capture_output=True, text=True, timeout=5 + ["sysctl", "-n", "hw.memsize"], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: mem_bytes = int(result.stdout.strip()) @@ -160,7 +161,9 @@ def _detect_nvidia_gpu(self) -> bool: try: result = subprocess.run( ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], - capture_output=True, text=True, timeout=10 + capture_output=True, + text=True, + timeout=10, ) return result.returncode == 0 and result.stdout.strip() != "" except Exception: @@ -171,7 +174,9 @@ def _get_nvidia_vram(self) -> int: try: result = subprocess.run( ["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits"], - capture_output=True, text=True, timeout=10 + capture_output=True, + text=True, + timeout=10, ) if result.returncode == 0: return int(result.stdout.strip().split("\n")[0]) @@ -223,7 +228,7 @@ def _benchmark_cpu(self) -> BenchmarkResult: score=score, raw_value=round(avg_time * 1000, 2), unit="ms", - description="Matrix computation speed" + description="Matrix computation speed", ) def _benchmark_memory(self) -> BenchmarkResult: @@ -250,7 +255,7 @@ def _benchmark_memory(self) -> BenchmarkResult: # Calculate approximate bandwidth (bytes per second) bytes_processed = size * 8 * 2 # 8 bytes per int, 2 operations - bandwidth_gbps = (bytes_processed / avg_time) / (1024 ** 3) + bandwidth_gbps = (bytes_processed / avg_time) / (1024**3) # Score based on bandwidth # Baseline: 10 GB/s = 50, 50 GB/s = 100, 1 GB/s = 10 @@ -267,7 +272,7 @@ def _benchmark_memory(self) -> BenchmarkResult: score=score, raw_value=round(bandwidth_gbps, 2), unit="GB/s", - description="Memory throughput" + description="Memory throughput", ) def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: @@ -298,7 +303,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=score, raw_value=vram_mb, unit="MB", - description="NVIDIA GPU VRAM" + description="NVIDIA GPU VRAM", ) elif system_info.get("has_apple_silicon"): @@ -320,7 +325,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=score, raw_value=int(ram_gb * 1024), unit="MB (unified)", - description="Apple Silicon unified memory" + description="Apple Silicon unified memory", ) else: @@ -330,7 +335,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=15, raw_value=0, unit="MB", - description="No dedicated GPU detected" + description="No dedicated GPU detected", ) def _benchmark_inference_simulation(self) -> BenchmarkResult: @@ -348,9 +353,11 @@ def _benchmark_inference_simulation(self) -> BenchmarkResult: # Simulate embedding lookup (string hashing) embeddings = [hash(token) % 10000 for token in tokens] # Simulate attention (nested loops) - attention = sum(embeddings[i] * embeddings[j] - for i in range(min(50, len(embeddings))) - for j in range(min(50, len(embeddings)))) + attention = sum( + embeddings[i] * embeddings[j] + for i in range(min(50, len(embeddings))) + for j in range(min(50, len(embeddings))) + ) _ = attention elapsed = time.perf_counter() - start @@ -372,7 +379,7 @@ def _benchmark_inference_simulation(self) -> BenchmarkResult: score=score, raw_value=round(tokens_per_sec / 1000, 2), unit="K tok/s", - description="Simulated inference throughput" + description="Simulated inference throughput", ) def _benchmark_token_generation(self) -> BenchmarkResult: @@ -390,8 +397,10 @@ def _benchmark_token_generation(self) -> BenchmarkResult: context = [0] * 10 for _ in range(sequence_length): # Simulate softmax over vocabulary - logits = [(hash((i, tuple(context[-10:]))) % 1000) / 1000 - for i in range(min(1000, vocab_size))] + logits = [ + (hash((i, tuple(context[-10:]))) % 1000) / 1000 + for i in range(min(1000, vocab_size)) + ] next_token = max(range(len(logits)), key=lambda i: logits[i]) generated.append(next_token) context.append(next_token) @@ -415,10 +424,10 @@ def _benchmark_token_generation(self) -> BenchmarkResult: score=score, raw_value=round(tokens_per_sec, 1), unit="tok/s", - description="Simulated generation speed" + description="Simulated generation speed", ) - def _calculate_overall_score(self, results: List[BenchmarkResult]) -> Tuple[int, str]: + def _calculate_overall_score(self, results: list[BenchmarkResult]) -> tuple[int, str]: """ Calculate overall score and rating. @@ -465,7 +474,7 @@ def _calculate_overall_score(self, results: List[BenchmarkResult]) -> Tuple[int, def _get_model_recommendations( self, system_info: dict, overall_score: int - ) -> Tuple[List[str], List[str], str]: + ) -> tuple[list[str], list[str], str]: """ Get model recommendations based on system capabilities. @@ -579,8 +588,9 @@ def run(self, save_history: bool = True) -> BenchmarkReport: report.overall_score, report.rating = self._calculate_overall_score(report.results) # Get model recommendations - report.can_run, report.needs_upgrade, report.upgrade_suggestion = \ + report.can_run, report.needs_upgrade, report.upgrade_suggestion = ( self._get_model_recommendations(report.system_info, report.overall_score) + ) # Save to history if save_history: @@ -633,11 +643,7 @@ def display_report(self, report: BenchmarkReport): else: score_str = f"[red]{result.score}/100[/red]" - table.add_row( - result.name, - score_str, - f"{result.raw_value} {result.unit}" - ) + table.add_row(result.name, score_str, f"{result.raw_value} {result.unit}") console.print(table) console.print() @@ -650,12 +656,16 @@ def display_report(self, report: BenchmarkReport): else: score_color = "red" - score_content = f"[bold {score_color}]{report.overall_score}/100[/bold {score_color}] ({report.rating})" - console.print(Panel( - f"[bold]OVERALL SCORE:[/bold] {score_content}", - border_style="cyan", - box=box.ROUNDED, - )) + score_content = ( + f"[bold {score_color}]{report.overall_score}/100[/bold {score_color}] ({report.rating})" + ) + console.print( + Panel( + f"[bold]OVERALL SCORE:[/bold] {score_content}", + border_style="cyan", + box=box.ROUNDED, + ) + ) console.print() # Model recommendations diff --git a/cortex/branding.py b/cortex/branding.py index de412be3b..f3ed47b32 100644 --- a/cortex/branding.py +++ b/cortex/branding.py @@ -11,7 +11,7 @@ - Consistent visual language """ -from typing import List, Optional, Tuple +from typing import Optional from rich import box from rich.console import Console @@ -141,8 +141,8 @@ def show_goodbye(): def cx_box( content: str, - title: Optional[str] = None, - subtitle: Optional[str] = None, + title: str | None = None, + subtitle: str | None = None, status: str = "info", ) -> None: """ @@ -175,7 +175,7 @@ def cx_box( def cx_status_box( title: str, - items: List[Tuple[str, str, str]], + items: list[tuple[str, str, str]], ) -> None: """ Print a status box with aligned key-value pairs. @@ -221,10 +221,10 @@ def cx_status_box( def cx_table( - headers: List[str], - rows: List[List[str]], - title: Optional[str] = None, - row_styles: Optional[List[str]] = None, + headers: list[str], + rows: list[list[str]], + title: str | None = None, + row_styles: list[str] | None = None, ) -> None: """ Print a formatted table with Cortex styling. @@ -255,7 +255,7 @@ def cx_table( def cx_package_table( - packages: List[Tuple[str, str, str]], + packages: list[tuple[str, str, str]], title: str = "Packages", ) -> None: """ @@ -293,7 +293,7 @@ def cx_package_table( console.print(table) -def cx_divider(title: Optional[str] = None) -> None: +def cx_divider(title: str | None = None) -> None: """ Print a horizontal divider with optional title. @@ -318,7 +318,9 @@ def cx_error(message: str) -> None: def cx_warning(message: str) -> None: """Print a warning message with warning icon.""" - console.print(f"[{CORTEX_WARNING}]⚠[/{CORTEX_WARNING}] [{CORTEX_WARNING}]{message}[/{CORTEX_WARNING}]") + console.print( + f"[{CORTEX_WARNING}]⚠[/{CORTEX_WARNING}] [{CORTEX_WARNING}]{message}[/{CORTEX_WARNING}]" + ) def cx_info(message: str) -> None: diff --git a/cortex/cli.py b/cortex/cli.py index 1f3d15967..9482fea7a 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -14,9 +14,6 @@ from cortex.ask import AskHandler from cortex.branding import VERSION, console, cx_header, cx_print, show_banner from cortex.coordinator import InstallationCoordinator, InstallationStep, StepStatus -from cortex.update_checker import UpdateChannel, should_notify_update -from cortex.updater import Updater, UpdateStatus -from cortex.version_manager import get_version_string from cortex.demo import run_demo from cortex.dependency_importer import ( DependencyImporter, @@ -31,7 +28,10 @@ from cortex.notification_manager import NotificationManager from cortex.role_manager import RoleManager from cortex.stack_manager import StackManager +from cortex.update_checker import UpdateChannel, should_notify_update +from cortex.updater import Updater, UpdateStatus from cortex.validators import validate_api_key, validate_install_request +from cortex.version_manager import get_version_string if TYPE_CHECKING: from cortex.shell_env_analyzer import ShellEnvironmentAnalyzer @@ -1218,11 +1218,11 @@ def update(self, args: argparse.Namespace) -> int: "success", ) console.print() - console.print(f"[bold]Release notes:[/bold]") + console.print("[bold]Release notes:[/bold]") console.print(result.latest_release.release_notes_summary) console.print() cx_print( - f"Run [bold]cortex update install[/bold] to upgrade", + "Run [bold]cortex update install[/bold] to upgrade", "info", ) else: @@ -1262,9 +1262,7 @@ def progress_callback(message: str, percent: float) -> None: "success", ) if result.duration_seconds: - console.print( - f"[dim]Completed in {result.duration_seconds:.1f}s[/dim]" - ) + console.print(f"[dim]Completed in {result.duration_seconds:.1f}s[/dim]") elif result.status == UpdateStatus.PENDING: # Dry run cx_print( @@ -2544,9 +2542,7 @@ def main(): f"[cyan]🔔 Cortex update available:[/cyan] " f"[green]{update_release.version}[/green]" ) - console.print( - f" [dim]Run 'cortex update' to upgrade[/dim]" - ) + console.print(" [dim]Run 'cortex update' to upgrade[/dim]") console.print() except Exception: pass # Don't block CLI on update check failures @@ -2600,7 +2596,7 @@ def main(): nargs="?", default="status", choices=["status", "diagnose", "deps"], - help="Action: status (default), diagnose, deps" + help="Action: status (default), diagnose, deps", ) systemd_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") @@ -2611,9 +2607,11 @@ def main(): nargs="?", default="status", choices=["status", "modes", "switch", "apps"], - help="Action: status (default), modes, switch, apps" + help="Action: status (default), modes, switch, apps", + ) + gpu_parser.add_argument( + "mode", nargs="?", help="Mode for switch action (integrated/hybrid/nvidia)" ) - gpu_parser.add_argument("mode", nargs="?", help="Mode for switch action (integrated/hybrid/nvidia)") gpu_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") # Printer/Scanner setup command @@ -2623,7 +2621,7 @@ def main(): nargs="?", default="status", choices=["status", "detect"], - help="Action: status (default), detect" + help="Action: status (default), detect", ) printer_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") @@ -3051,7 +3049,8 @@ def main(): help="Action to perform (default: status)", ) wifi_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3078,7 +3077,8 @@ def main(): help="Truncation mode for large input (default: middle)", ) stdin_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3098,7 +3098,8 @@ def main(): help="Package constraints (format: pkg:constraint:source)", ) deps_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3113,7 +3114,8 @@ def main(): help="Action to perform (default: check)", ) health_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3148,18 +3150,17 @@ def main(): return cli.systemd( args.service, action=getattr(args, "action", "status"), - verbose=getattr(args, "verbose", False) + verbose=getattr(args, "verbose", False), ) elif args.command == "gpu": return cli.gpu( action=getattr(args, "action", "status"), mode=getattr(args, "mode", None), - verbose=getattr(args, "verbose", False) + verbose=getattr(args, "verbose", False), ) elif args.command == "printer": return cli.printer( - action=getattr(args, "action", "status"), - verbose=getattr(args, "verbose", False) + action=getattr(args, "action", "status"), verbose=getattr(args, "verbose", False) ) elif args.command == "ask": return cli.ask(args.question) @@ -3193,25 +3194,30 @@ def main(): return cli.env(args) elif args.command == "upgrade": from cortex.licensing import open_upgrade_page + open_upgrade_page() return 0 elif args.command == "license": from cortex.licensing import show_license_status + show_license_status() return 0 elif args.command == "activate": from cortex.licensing import activate_license + return 0 if activate_license(args.license_key) else 1 elif args.command == "update": return cli.update(args) elif args.command == "wifi": from cortex.wifi_driver import run_wifi_driver + return run_wifi_driver( action=getattr(args, "action", "status"), verbose=getattr(args, "verbose", False), ) elif args.command == "stdin": from cortex.stdin_handler import run_stdin_handler + return run_stdin_handler( action=getattr(args, "action", "info"), max_lines=getattr(args, "max_lines", 1000), @@ -3220,6 +3226,7 @@ def main(): ) elif args.command == "deps": from cortex.semver_resolver import run_semver_resolver + return run_semver_resolver( action=getattr(args, "action", "analyze"), packages=getattr(args, "packages", None), @@ -3227,6 +3234,7 @@ def main(): ) elif args.command == "health": from cortex.health_score import run_health_check + return run_health_check( action=getattr(args, "action", "check"), verbose=getattr(args, "verbose", False), diff --git a/cortex/gpu_manager.py b/cortex/gpu_manager.py index ee6245fdc..171708ddf 100644 --- a/cortex/gpu_manager.py +++ b/cortex/gpu_manager.py @@ -13,14 +13,14 @@ from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Optional from rich import box from rich.console import Console from rich.panel import Panel from rich.table import Table -from cortex.branding import console, cx_print, cx_header, CORTEX_CYAN +from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print class GPUMode(Enum): @@ -61,8 +61,8 @@ class GPUState: """Current GPU system state.""" mode: GPUMode = GPUMode.UNKNOWN - devices: List[GPUDevice] = field(default_factory=list) - active_gpu: Optional[GPUDevice] = None + devices: list[GPUDevice] = field(default_factory=list) + active_gpu: GPUDevice | None = None prime_profile: str = "" render_offload_available: bool = False power_management: str = "" @@ -83,7 +83,7 @@ class AppGPUConfig: name: str executable: str gpu: GPUVendor - env_vars: Dict[str, str] = field(default_factory=dict) + env_vars: dict[str, str] = field(default_factory=dict) # Battery impact estimates (hours difference) @@ -126,24 +126,19 @@ class HybridGPUManager: def __init__(self, verbose: bool = False): self.verbose = verbose - self._state: Optional[GPUState] = None + self._state: GPUState | None = None - def _run_command(self, cmd: List[str], timeout: int = 10) -> Tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 10) -> tuple[int, str, str]: """Run a command and return (returncode, stdout, stderr).""" try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout - ) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", f"Command not found: {cmd[0]}" except subprocess.TimeoutExpired: return 1, "", "Command timed out" - def detect_gpus(self) -> List[GPUDevice]: + def detect_gpus(self) -> list[GPUDevice]: """Detect all GPU devices in the system.""" devices = [] @@ -172,7 +167,7 @@ def detect_gpus(self) -> List[GPUDevice]: return devices - def _parse_lspci_line(self, line: str) -> Optional[GPUDevice]: + def _parse_lspci_line(self, line: str) -> GPUDevice | None: """Parse an lspci output line for GPU info.""" line_lower = line.lower() @@ -200,13 +195,15 @@ def _parse_lspci_line(self, line: str) -> Optional[GPUDevice]: pci_id=pci_id, ) - def _detect_nvidia_gpu(self) -> Optional[GPUDevice]: + def _detect_nvidia_gpu(self) -> GPUDevice | None: """Detect NVIDIA GPU with detailed info.""" - returncode, stdout, _ = self._run_command([ - "nvidia-smi", - "--query-gpu=name,memory.total,power.draw", - "--format=csv,noheader,nounits" - ]) + returncode, stdout, _ = self._run_command( + [ + "nvidia-smi", + "--query-gpu=name,memory.total,power.draw", + "--format=csv,noheader,nounits", + ] + ) if returncode != 0 or not stdout.strip(): return None @@ -216,9 +213,9 @@ def _detect_nvidia_gpu(self) -> Optional[GPUDevice]: memory = int(float(parts[1].strip())) if len(parts) > 1 else 0 # Check power state - power_returncode, power_stdout, _ = self._run_command([ - "cat", "/sys/bus/pci/devices/0000:01:00.0/power/runtime_status" - ]) + power_returncode, power_stdout, _ = self._run_command( + ["cat", "/sys/bus/pci/devices/0000:01:00.0/power/runtime_status"] + ) power_state = power_stdout.strip() if power_returncode == 0 else "unknown" return GPUDevice( @@ -278,10 +275,15 @@ def get_state(self, refresh: bool = False) -> GPUState: # Find active GPU for device in state.devices: - if device.is_active or (state.mode == GPUMode.NVIDIA and device.vendor == GPUVendor.NVIDIA): + if device.is_active or ( + state.mode == GPUMode.NVIDIA and device.vendor == GPUVendor.NVIDIA + ): state.active_gpu = device break - elif state.mode == GPUMode.INTEGRATED and device.vendor in [GPUVendor.INTEL, GPUVendor.AMD]: + elif state.mode == GPUMode.INTEGRATED and device.vendor in [ + GPUVendor.INTEL, + GPUVendor.AMD, + ]: state.active_gpu = device break @@ -292,7 +294,7 @@ def get_state(self, refresh: bool = False) -> GPUState: self._state = state return state - def switch_mode(self, mode: GPUMode, apply: bool = False) -> Tuple[bool, str, Optional[str]]: + def switch_mode(self, mode: GPUMode, apply: bool = False) -> tuple[bool, str, str | None]: """ Switch GPU mode. @@ -347,7 +349,11 @@ def switch_mode(self, mode: GPUMode, apply: bool = False) -> Tuple[bool, str, Op command = f"sudo system76-power graphics {mode_map[mode]}" if not command: - return False, "No GPU switching tool found. Install prime-select, envycontrol, or system76-power.", None + return ( + False, + "No GPU switching tool found. Install prime-select, envycontrol, or system76-power.", + None, + ) if apply: # Actually run the command (would need sudo) @@ -385,7 +391,7 @@ def get_app_launch_command(self, app: str, use_nvidia: bool = True) -> str: # Use integrated GPU return f"DRI_PRIME=0 {app}" - def get_battery_estimate(self, mode: GPUMode) -> Dict[str, str]: + def get_battery_estimate(self, mode: GPUMode) -> dict[str, str]: """Get battery impact estimate for a mode.""" return BATTERY_IMPACT.get(mode, {"description": "Unknown", "impact": "Unknown"}) @@ -444,12 +450,14 @@ def display_status(self): [dim]{mode_info['description']}[/dim] Battery Impact: {mode_info['impact']} """ - console.print(Panel( - mode_panel, - title="[bold cyan]GPU Mode[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + mode_panel, + title="[bold cyan]GPU Mode[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) if state.is_hybrid_system: console.print() @@ -517,11 +525,7 @@ def display_app_recommendations(self): console.print(table) -def run_gpu_manager( - action: str = "status", - mode: Optional[str] = None, - verbose: bool = False -) -> int: +def run_gpu_manager(action: str = "status", mode: str | None = None, verbose: bool = False) -> int: """ Main entry point for cortex gpu command. diff --git a/cortex/health_score.py b/cortex/health_score.py index de12e7c25..2e68c9bd8 100644 --- a/cortex/health_score.py +++ b/cortex/health_score.py @@ -10,11 +10,12 @@ import os import subprocess import time +from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime from enum import Enum from pathlib import Path -from typing import Callable, Optional +from typing import Optional from rich.console import Console from rich.panel import Panel @@ -142,9 +143,7 @@ def __init__(self, verbose: bool = False): self.verbose = verbose self.history_path = Path.home() / ".cortex" / "health_history.json" - def _run_command( - self, cmd: list[str], timeout: int = 30 - ) -> tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return exit code, stdout, stderr.""" try: result = subprocess.run( @@ -308,9 +307,7 @@ def check_security(self) -> HealthFactor: pass # Check for unattended upgrades - code, _, _ = self._run_command( - ["dpkg", "-l", "unattended-upgrades"] - ) + code, _, _ = self._run_command(["dpkg", "-l", "unattended-upgrades"]) if code != 0: issues.append("Automatic updates not configured") score -= 10 @@ -477,16 +474,13 @@ def save_history(self, report: HealthReport): try: with open(self.history_path) as f: history = json.load(f) - except (json.JSONDecodeError, IOError): + except (OSError, json.JSONDecodeError): history = [] entry = { "timestamp": report.timestamp.isoformat(), "overall_score": report.overall_score, - "factors": { - f.name: {"score": f.score, "details": f.details} - for f in report.factors - }, + "factors": {f.name: {"score": f.score, "details": f.details} for f in report.factors}, } history.append(entry) @@ -497,7 +491,7 @@ def save_history(self, report: HealthReport): try: with open(self.history_path, "w") as f: json.dump(history, f, indent=2) - except IOError: + except OSError: pass def load_history(self) -> list[dict]: @@ -508,7 +502,7 @@ def load_history(self) -> list[dict]: try: with open(self.history_path) as f: return json.load(f) - except (json.JSONDecodeError, IOError): + except (OSError, json.JSONDecodeError): return [] def display_report(self, report: HealthReport): @@ -587,9 +581,7 @@ def display_history(self): else: trend = "→" - score_color = ( - "green" if score >= 75 else "yellow" if score >= 50 else "red" - ) + score_color = "green" if score >= 75 else "yellow" if score >= 50 else "red" table.add_row( ts.strftime("%Y-%m-%d %H:%M"), diff --git a/cortex/licensing.py b/cortex/licensing.py index 9fef81734..714832f14 100644 --- a/cortex/licensing.py +++ b/cortex/licensing.py @@ -2,9 +2,9 @@ import json import webbrowser +from datetime import datetime, timezone from pathlib import Path from typing import Optional -from datetime import datetime, timezone import httpx @@ -43,7 +43,6 @@ def level(tier: str) -> int: "parallel_ops": FeatureTier.PRO, "priority_support": FeatureTier.PRO, "usage_analytics": FeatureTier.PRO, - # Enterprise features ($99/month) "sso": FeatureTier.ENTERPRISE, "ldap": FeatureTier.ENTERPRISE, @@ -81,9 +80,9 @@ def __init__( self, tier: str = FeatureTier.COMMUNITY, valid: bool = True, - expires: Optional[datetime] = None, - organization: Optional[str] = None, - email: Optional[str] = None, + expires: datetime | None = None, + organization: str | None = None, + email: str | None = None, ): self.tier = tier self.valid = valid @@ -107,7 +106,7 @@ def days_remaining(self) -> int: return max(0, delta.days) -_cached_license: Optional[LicenseInfo] = None +_cached_license: LicenseInfo | None = None def get_license_info() -> LicenseInfo: @@ -183,12 +182,15 @@ def require_feature(feature_name: str): Raises: FeatureNotAvailableError: If feature not available """ + def decorator(func): def wrapper(*args, **kwargs): if not check_feature(feature_name): raise FeatureNotAvailableError(feature_name) return func(*args, **kwargs) + return wrapper + return decorator @@ -199,7 +201,8 @@ def show_upgrade_prompt(feature: str, required_tier: str) -> None: price = "$20" if required_tier == FeatureTier.PRO else "$99" - print(f""" + print( + f""" ┌─────────────────────────────────────────────────────────┐ │ ⚡ UPGRADE REQUIRED │ ├─────────────────────────────────────────────────────────┤ @@ -213,7 +216,8 @@ def show_upgrade_prompt(feature: str, required_tier: str) -> None: │ 🌐 {PRICING_URL} │ │ └─────────────────────────────────────────────────────────┘ -""") +""" + ) def show_license_status() -> None: @@ -226,12 +230,14 @@ def show_license_status() -> None: FeatureTier.ENTERPRISE: "yellow", } - print(f""" + print( + f""" ┌─────────────────────────────────────────────────────────┐ │ CORTEX LICENSE STATUS │ ├─────────────────────────────────────────────────────────┤ │ Tier: {info.tier.upper():12} │ -│ Status: {"ACTIVE" if info.valid else "EXPIRED":12} │""") +│ Status: {"ACTIVE" if info.valid else "EXPIRED":12} │""" + ) if info.organization: print(f"│ Organization: {info.organization[:12]:12} │") @@ -251,7 +257,7 @@ def show_license_status() -> None: print(f" {icon} {name}") if info.tier == FeatureTier.COMMUNITY: - print(f"\n 💡 Upgrade to Pro for just $20/month: cortex upgrade") + print("\n 💡 Upgrade to Pro for just $20/month: cortex upgrade") def activate_license(license_key: str) -> bool: @@ -280,19 +286,23 @@ def activate_license(license_key: str) -> bool: if data.get("success"): # Save license locally LICENSE_FILE.parent.mkdir(parents=True, exist_ok=True) - LICENSE_FILE.write_text(json.dumps({ - "key": license_key, - "tier": data["tier"], - "valid": True, - "expires": data.get("expires"), - "organization": data.get("organization"), - "email": data.get("email"), - })) + LICENSE_FILE.write_text( + json.dumps( + { + "key": license_key, + "tier": data["tier"], + "valid": True, + "expires": data.get("expires"), + "organization": data.get("organization"), + "email": data.get("email"), + } + ) + ) # Clear cache _cached_license = None - print(f"\n ✅ License activated successfully!") + print("\n ✅ License activated successfully!") print(f" Tier: {data['tier'].upper()}") if data.get("organization"): print(f" Organization: {data['organization']}") @@ -303,7 +313,7 @@ def activate_license(license_key: str) -> bool: return False except httpx.HTTPError as e: - print(f"\n ❌ Activation failed: Could not reach license server") + print("\n ❌ Activation failed: Could not reach license server") return False @@ -316,6 +326,7 @@ def open_upgrade_page() -> None: def _get_hostname() -> str: """Get system hostname.""" import platform + return platform.node() diff --git a/cortex/output_formatter.py b/cortex/output_formatter.py index e4a868be5..b015619ba 100644 --- a/cortex/output_formatter.py +++ b/cortex/output_formatter.py @@ -7,10 +7,11 @@ Issue: #242 """ +from collections.abc import Generator from contextlib import contextmanager from dataclasses import dataclass, field from enum import Enum -from typing import Any, Generator, List, Optional, Tuple +from typing import Any, Optional from rich import box from rich.console import Console, Group @@ -82,7 +83,7 @@ class TableColumn: header: str style: str = "cyan" justify: str = "left" - width: Optional[int] = None + width: int | None = None no_wrap: bool = False @@ -97,11 +98,11 @@ class StatusInfo: def format_box( content: str, - title: Optional[str] = None, - subtitle: Optional[str] = None, + title: str | None = None, + subtitle: str | None = None, style: OutputStyle = OutputStyle.DEFAULT, border_style: str = "cyan", - padding: Tuple[int, int] = (1, 2), + padding: tuple[int, int] = (1, 2), expand: bool = False, ) -> Panel: """ @@ -144,7 +145,7 @@ def format_box( def format_status_box( title: str, - items: List[StatusInfo], + items: list[StatusInfo], border_style: str = "cyan", ) -> Panel: """ @@ -196,12 +197,12 @@ def format_status_box( def format_table( - columns: List[TableColumn], - rows: List[List[str]], - title: Optional[str] = None, + columns: list[TableColumn], + rows: list[list[str]], + title: str | None = None, show_header: bool = True, show_lines: bool = False, - row_styles: Optional[List[str]] = None, + row_styles: list[str] | None = None, ) -> Table: """ Create a formatted table. @@ -244,7 +245,7 @@ def format_table( def format_package_table( - packages: List[Tuple[str, str, str]], + packages: list[tuple[str, str, str]], title: str = "Packages", ) -> Table: """ @@ -270,7 +271,7 @@ def format_package_table( def format_dependency_tree( package: str, dependencies: dict, - title: Optional[str] = None, + title: str | None = None, ) -> Tree: """ Create a formatted dependency tree. @@ -306,8 +307,8 @@ def add_deps(parent_tree: Tree, pkg: str, visited: set): @contextmanager def spinner_context( message: str, - success_message: Optional[str] = None, - error_message: Optional[str] = None, + success_message: str | None = None, + error_message: str | None = None, spinner_type: str = "dots", ) -> Generator[Status, None, None]: """ @@ -353,14 +354,14 @@ class ProgressTracker: def __init__( self, description: str, - total: Optional[int] = None, + total: int | None = None, show_speed: bool = False, ): self.description = description self.total = total self.show_speed = show_speed - self._progress: Optional[Progress] = None - self._task_id: Optional[TaskID] = None + self._progress: Progress | None = None + self._task_id: TaskID | None = None def __enter__(self) -> "ProgressTracker": columns = [ @@ -387,7 +388,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): if self._progress: self._progress.stop() - def update(self, description: Optional[str] = None, advance: int = 0): + def update(self, description: str | None = None, advance: int = 0): """Update progress description and/or advance by given amount.""" if self._progress and self._task_id is not None: self._progress.update( @@ -420,11 +421,11 @@ class MultiStepProgress: progress.complete_step(step) """ - def __init__(self, steps: List[str], title: str = "Operation Progress"): + def __init__(self, steps: list[str], title: str = "Operation Progress"): self.steps = steps self.title = title - self.step_status: dict = {step: "pending" for step in steps} - self._live: Optional[Live] = None + self.step_status: dict = dict.fromkeys(steps, "pending") + self._live: Live | None = None def __enter__(self) -> "MultiStepProgress": self._live = Live(self._render(), console=console, refresh_per_second=10) @@ -523,17 +524,17 @@ def print_box(content: str, **kwargs): console.print(format_box(content, **kwargs)) -def print_status_box(title: str, items: List[StatusInfo], **kwargs): +def print_status_box(title: str, items: list[StatusInfo], **kwargs): """Print a status box with key-value pairs.""" console.print(format_status_box(title, items, **kwargs)) -def print_table(columns: List[TableColumn], rows: List[List[str]], **kwargs): +def print_table(columns: list[TableColumn], rows: list[list[str]], **kwargs): """Print a formatted table.""" console.print(format_table(columns, rows, **kwargs)) -def print_divider(title: Optional[str] = None, style: str = "cyan"): +def print_divider(title: str | None = None, style: str = "cyan"): """Print a horizontal divider with optional title.""" if title: console.print(f"\n[bold {style}]━━━ {title} ━━━[/bold {style}]\n") diff --git a/cortex/printer_setup.py b/cortex/printer_setup.py index 68caafe0f..5b1733a67 100644 --- a/cortex/printer_setup.py +++ b/cortex/printer_setup.py @@ -11,14 +11,14 @@ from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Dict, List, Optional, Tuple +from typing import Optional from rich import box from rich.console import Console from rich.panel import Panel from rich.table import Table -from cortex.branding import console, cx_print, cx_header, CORTEX_CYAN +from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print class DeviceType(Enum): @@ -103,15 +103,10 @@ def __init__(self, verbose: bool = False): self._cups_available = self._check_cups() self._sane_available = self._check_sane() - def _run_command(self, cmd: List[str], timeout: int = 30) -> Tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return (returncode, stdout, stderr).""" try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout - ) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", f"Command not found: {cmd[0]}" @@ -128,7 +123,7 @@ def _check_sane(self) -> bool: returncode, _, _ = self._run_command(["which", "scanimage"]) return returncode == 0 - def detect_usb_printers(self) -> List[PrinterDevice]: + def detect_usb_printers(self) -> list[PrinterDevice]: """Detect USB-connected printers.""" devices = [] @@ -161,17 +156,19 @@ def detect_usb_printers(self) -> List[PrinterDevice]: else: device_type = DeviceType.PRINTER - devices.append(PrinterDevice( - name=name, - device_type=device_type, - connection=ConnectionType.USB, - vendor=vendor, - usb_id=usb_id, - )) + devices.append( + PrinterDevice( + name=name, + device_type=device_type, + connection=ConnectionType.USB, + vendor=vendor, + usb_id=usb_id, + ) + ) return devices - def detect_network_printers(self) -> List[PrinterDevice]: + def detect_network_printers(self) -> list[PrinterDevice]: """Detect network printers using CUPS and DNS-SD.""" devices = [] @@ -188,17 +185,19 @@ def detect_network_printers(self) -> List[PrinterDevice]: uri = parts[1] name = uri.split("/")[-1] if "/" in uri else uri - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.PRINTER, - connection=ConnectionType.NETWORK, - uri=uri, - vendor=self._detect_vendor(name), - )) + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.PRINTER, + connection=ConnectionType.NETWORK, + uri=uri, + vendor=self._detect_vendor(name), + ) + ) return devices - def detect_configured_printers(self) -> List[PrinterDevice]: + def detect_configured_printers(self) -> list[PrinterDevice]: """Get list of already configured printers.""" devices = [] @@ -221,20 +220,30 @@ def detect_configured_printers(self) -> List[PrinterDevice]: parts = line.split() if len(parts) >= 2: name = parts[1] - state = "idle" if "is idle" in line else "printing" if "printing" in line else "disabled" if "disabled" in line else "unknown" - - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.PRINTER, - connection=ConnectionType.UNKNOWN, - is_configured=True, - is_default=name == default_printer, - state=state, - )) + state = ( + "idle" + if "is idle" in line + else ( + "printing" + if "printing" in line + else "disabled" if "disabled" in line else "unknown" + ) + ) + + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.PRINTER, + connection=ConnectionType.UNKNOWN, + is_configured=True, + is_default=name == default_printer, + state=state, + ) + ) return devices - def detect_scanners(self) -> List[PrinterDevice]: + def detect_scanners(self) -> list[PrinterDevice]: """Detect scanners using SANE.""" devices = [] @@ -256,14 +265,16 @@ def detect_scanners(self) -> List[PrinterDevice]: if "net:" in uri or "airscan:" in uri: connection = ConnectionType.NETWORK - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.SCANNER, - connection=connection, - uri=uri, - vendor=self._detect_vendor(name), - is_configured=True, - )) + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.SCANNER, + connection=connection, + uri=uri, + vendor=self._detect_vendor(name), + is_configured=True, + ) + ) return devices @@ -288,7 +299,7 @@ def _detect_vendor(self, name: str) -> str: return "generic" - def find_driver(self, device: PrinterDevice) -> Optional[DriverInfo]: + def find_driver(self, device: PrinterDevice) -> DriverInfo | None: """Find the best driver for a device.""" if not self._cups_available: return None @@ -319,7 +330,7 @@ def find_driver(self, device: PrinterDevice) -> Optional[DriverInfo]: source="cups-generic", ) - def get_driver_packages(self, device: PrinterDevice) -> List[str]: + def get_driver_packages(self, device: PrinterDevice) -> list[str]: """Get recommended driver packages for a device.""" packages = [] @@ -336,9 +347,9 @@ def get_driver_packages(self, device: PrinterDevice) -> List[str]: def setup_printer( self, device: PrinterDevice, - driver: Optional[DriverInfo] = None, + driver: DriverInfo | None = None, make_default: bool = False, - ) -> Tuple[bool, str]: + ) -> tuple[bool, str]: """ Set up a printer with CUPS. @@ -360,7 +371,7 @@ def setup_printer( return False, f"Could not find driver for {device.name}" # Generate a safe printer name - printer_name = re.sub(r'[^a-zA-Z0-9_-]', '_', device.name)[:30] + printer_name = re.sub(r"[^a-zA-Z0-9_-]", "_", device.name)[:30] # Determine URI uri = device.uri @@ -379,9 +390,12 @@ def setup_printer( # Add printer cmd = [ "lpadmin", - "-p", printer_name, - "-v", uri, - "-m", driver.ppd_path, + "-p", + printer_name, + "-v", + uri, + "-m", + driver.ppd_path, "-E", # Enable ] @@ -395,23 +409,22 @@ def setup_printer( return True, f"Printer '{printer_name}' configured successfully" - def test_print(self, printer_name: str) -> Tuple[bool, str]: + def test_print(self, printer_name: str) -> tuple[bool, str]: """Send a test page to a printer.""" if not self._cups_available: return False, "CUPS is not installed" # Use CUPS test page - returncode, _, stderr = self._run_command([ - "lp", "-d", printer_name, - "/usr/share/cups/data/testprint" - ]) + returncode, _, stderr = self._run_command( + ["lp", "-d", printer_name, "/usr/share/cups/data/testprint"] + ) if returncode == 0: return True, "Test page sent to printer" else: return False, f"Failed to print test page: {stderr}" - def test_scan(self, scanner_uri: Optional[str] = None) -> Tuple[bool, str]: + def test_scan(self, scanner_uri: str | None = None) -> tuple[bool, str]: """Test scanner functionality.""" if not self._sane_available: return False, "SANE is not installed" @@ -454,11 +467,15 @@ def display_status(self): table.add_column("Default", style="green") for printer in configured: - status_color = "green" if printer.state == "idle" else "yellow" if printer.state == "printing" else "red" + status_color = ( + "green" + if printer.state == "idle" + else "yellow" if printer.state == "printing" else "red" + ) table.add_row( printer.name, f"[{status_color}]{printer.state}[/{status_color}]", - "✓" if printer.is_default else "" + "✓" if printer.is_default else "", ) console.print(table) @@ -469,7 +486,11 @@ def display_status(self): if usb_printers: console.print("[bold]Detected USB Devices:[/bold]") for printer in usb_printers: - icon = "🖨️" if printer.device_type == DeviceType.PRINTER else "📠" if printer.device_type == DeviceType.MULTIFUNCTION else "📷" + icon = ( + "🖨️" + if printer.device_type == DeviceType.PRINTER + else "📠" if printer.device_type == DeviceType.MULTIFUNCTION else "📷" + ) console.print(f" {icon} {printer.name} ({printer.vendor})") console.print() @@ -519,12 +540,14 @@ def display_setup_guide(self, device: PrinterDevice): if driver.recommended: content_lines.append("[green]✓ Recommended driver available[/green]") - console.print(Panel( - "\n".join(content_lines), - title="[bold cyan]Setup Information[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + "\n".join(content_lines), + title="[bold cyan]Setup Information[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) def run_printer_setup(action: str = "status", verbose: bool = False) -> int: diff --git a/cortex/semver_resolver.py b/cortex/semver_resolver.py index 38ff30b16..cec575f19 100644 --- a/cortex/semver_resolver.py +++ b/cortex/semver_resolver.py @@ -117,8 +117,8 @@ class VersionConstraint: raw: str constraint_type: ConstraintType - version: Optional[SemVer] = None - max_version: Optional[SemVer] = None # For range constraints + version: SemVer | None = None + max_version: SemVer | None = None # For range constraints def satisfies(self, version: SemVer) -> bool: """Check if a version satisfies this constraint.""" @@ -144,10 +144,7 @@ def satisfies(self, version: SemVer) -> bool: # ~1.2.3 means >=1.2.3 <1.3.0 if version < self.version: return False - return ( - version.major == self.version.major - and version.minor == self.version.minor - ) + return version.major == self.version.major and version.minor == self.version.minor elif self.constraint_type == ConstraintType.GREATER: return version > self.version @@ -184,7 +181,7 @@ class VersionConflict: package: str dependencies: list[Dependency] = field(default_factory=list) - resolved_version: Optional[SemVer] = None + resolved_version: SemVer | None = None @property def is_conflicting(self) -> bool: @@ -203,9 +200,7 @@ def is_conflicting(self) -> bool: return True return False - def _constraints_compatible( - self, c1: VersionConstraint, c2: VersionConstraint - ) -> bool: + def _constraints_compatible(self, c1: VersionConstraint, c2: VersionConstraint) -> bool: """Check if two constraints can be satisfied simultaneously.""" if c1.constraint_type == ConstraintType.ANY: return True @@ -261,7 +256,7 @@ def __init__(self, verbose: bool = False): self.dependencies: dict[str, list[Dependency]] = {} self.conflicts: list[VersionConflict] = [] - def parse_version(self, version_str: str) -> Optional[SemVer]: + def parse_version(self, version_str: str) -> SemVer | None: """Parse a semantic version string. Args: @@ -283,7 +278,7 @@ def parse_version(self, version_str: str) -> Optional[SemVer]: build=match.group("build") or "", ) - def parse_constraint(self, constraint_str: str) -> Optional[VersionConstraint]: + def parse_constraint(self, constraint_str: str) -> VersionConstraint | None: """Parse a version constraint string. Args: @@ -403,9 +398,7 @@ def parse_constraint(self, constraint_str: str) -> Optional[VersionConstraint]: return None - def add_dependency( - self, package: str, constraint_str: str, source: str = "" - ) -> bool: + def add_dependency(self, package: str, constraint_str: str, source: str = "") -> bool: """Add a dependency constraint. Args: @@ -446,9 +439,7 @@ def detect_conflicts(self) -> list[VersionConflict]: return self.conflicts - def suggest_resolutions( - self, conflict: VersionConflict - ) -> list[ResolutionStrategy]: + def suggest_resolutions(self, conflict: VersionConflict) -> list[ResolutionStrategy]: """Suggest resolution strategies for a conflict. Args: @@ -512,9 +503,7 @@ def suggest_resolutions( return strategies - def _find_common_version_strategy( - self, conflict: VersionConflict - ) -> Optional[ResolutionStrategy]: + def _find_common_version_strategy(self, conflict: VersionConflict) -> ResolutionStrategy | None: """Try to find a common version that satisfies all constraints.""" constraints = [d.constraint for d in conflict.dependencies] @@ -615,7 +604,7 @@ def display_resolutions(self, conflict: VersionConflict): def run_semver_resolver( action: str = "analyze", - packages: Optional[list[str]] = None, + packages: list[str] | None = None, verbose: bool = False, ) -> int: """Run the semantic version resolver. @@ -707,9 +696,7 @@ def run_semver_resolver( return 1 if constraint.satisfies(version): - console.print( - f"[green]Version {version} satisfies constraint {constraint_str}[/green]" - ) + console.print(f"[green]Version {version} satisfies constraint {constraint_str}[/green]") return 0 else: console.print( diff --git a/cortex/stdin_handler.py b/cortex/stdin_handler.py index 8c5ce4dd3..bc61749cc 100644 --- a/cortex/stdin_handler.py +++ b/cortex/stdin_handler.py @@ -104,7 +104,7 @@ def read_stdin(self) -> StdinData: try: content = sys.stdin.read() - except (IOError, OSError): + except OSError: return StdinData(content="", line_count=0, byte_count=0) lines = content.splitlines(keepends=True) @@ -141,11 +141,7 @@ def truncate(self, data: StdinData) -> StdinData: head = lines[:half] tail = lines[-half:] skipped = len(lines) - self.max_lines - truncated_lines = ( - head - + [f"\n... [{skipped} lines truncated] ...\n\n"] - + tail - ) + truncated_lines = head + [f"\n... [{skipped} lines truncated] ...\n\n"] + tail else: # SAMPLE step = max(1, len(lines) // self.max_lines) truncated_lines = lines[::step][: self.max_lines] @@ -155,9 +151,7 @@ def truncate(self, data: StdinData) -> StdinData: # Check byte limit content_bytes = content.encode("utf-8", errors="replace") if len(content_bytes) > self.max_bytes: - content = content_bytes[: self.max_bytes].decode( - "utf-8", errors="replace" - ) + content = content_bytes[: self.max_bytes].decode("utf-8", errors="replace") content += "\n... [truncated due to size limit] ..." new_lines = content.splitlines(keepends=True) @@ -230,21 +224,19 @@ def detect_content_type(content: str) -> str: return "json" # CSV - if "," in first_line and lines[0].count(",") == lines[1].count(",") if len(lines) > 1 else False: + if ( + "," in first_line and lines[0].count(",") == lines[1].count(",") + if len(lines) > 1 + else False + ): return "csv" # Docker/container logs - if any( - pattern in content - for pattern in ["container", "docker", "kubernetes", "pod"] - ): + if any(pattern in content for pattern in ["container", "docker", "kubernetes", "pod"]): return "container_log" # System logs - if any( - pattern in content - for pattern in ["systemd", "journald", "kernel", "syslog"] - ): + if any(pattern in content for pattern in ["systemd", "journald", "kernel", "syslog"]): return "system_log" return "text" @@ -309,7 +301,7 @@ def analyze_stdin( return result -def display_stdin_info(data: StdinData, analysis: Optional[dict] = None): +def display_stdin_info(data: StdinData, analysis: dict | None = None): """Display information about stdin data. Args: diff --git a/cortex/systemd_helper.py b/cortex/systemd_helper.py index 4a125e336..bc63775d0 100644 --- a/cortex/systemd_helper.py +++ b/cortex/systemd_helper.py @@ -13,7 +13,7 @@ from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Optional from rich import box from rich.console import Console @@ -21,7 +21,7 @@ from rich.table import Table from rich.tree import Tree -from cortex.branding import console, cx_print, cx_header, CORTEX_CYAN +from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print # Service state explanations in plain English SERVICE_STATE_EXPLANATIONS = { @@ -63,7 +63,10 @@ ("Verify dependencies are running", "systemctl list-dependencies {service}"), ], "signal": [ - ("Service was killed by a signal", "Check if OOM killer terminated it: dmesg | grep -i oom"), + ( + "Service was killed by a signal", + "Check if OOM killer terminated it: dmesg | grep -i oom", + ), ("Check resource limits", "systemctl show {service} | grep -i limit"), ], "timeout": [ @@ -75,8 +78,14 @@ ("Review application logs", "The application has a bug or invalid input."), ], "start-limit-hit": [ - ("Service crashed too many times", "Reset the failure count: systemctl reset-failed {service}"), - ("Fix the underlying issue", "Check logs before restarting: journalctl -u {service} -n 100"), + ( + "Service crashed too many times", + "Reset the failure count: systemctl reset-failed {service}", + ), + ( + "Fix the underlying issue", + "Check logs before restarting: journalctl -u {service} -n 100", + ), ], } @@ -99,17 +108,17 @@ class ServiceConfig: description: str exec_start: str service_type: ServiceType = ServiceType.SIMPLE - user: Optional[str] = None - group: Optional[str] = None - working_directory: Optional[str] = None - environment: Dict[str, str] = field(default_factory=dict) + user: str | None = None + group: str | None = None + working_directory: str | None = None + environment: dict[str, str] = field(default_factory=dict) restart: str = "on-failure" restart_sec: int = 5 - wants: List[str] = field(default_factory=list) - after: List[str] = field(default_factory=lambda: ["network.target"]) - wanted_by: List[str] = field(default_factory=lambda: ["multi-user.target"]) - exec_stop: Optional[str] = None - exec_reload: Optional[str] = None + wants: list[str] = field(default_factory=list) + after: list[str] = field(default_factory=lambda: ["network.target"]) + wanted_by: list[str] = field(default_factory=lambda: ["multi-user.target"]) + exec_stop: str | None = None + exec_reload: str | None = None timeout_start_sec: int = 90 timeout_stop_sec: int = 90 @@ -130,7 +139,7 @@ class ServiceStatus: since: str = "" result: str = "" fragment_path: str = "" - docs: List[str] = field(default_factory=list) + docs: list[str] = field(default_factory=list) class SystemdHelper: @@ -150,16 +159,11 @@ class SystemdHelper: def __init__(self, verbose: bool = False): self.verbose = verbose - def _run_systemctl(self, *args, capture: bool = True) -> Tuple[int, str, str]: + def _run_systemctl(self, *args, capture: bool = True) -> tuple[int, str, str]: """Run a systemctl command and return (returncode, stdout, stderr).""" cmd = ["systemctl"] + list(args) try: - result = subprocess.run( - cmd, - capture_output=capture, - text=True, - timeout=30 - ) + result = subprocess.run(cmd, capture_output=capture, text=True, timeout=30) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", "systemctl not found. Is systemd installed?" @@ -173,13 +177,13 @@ def _run_journalctl(self, service: str, lines: int = 50) -> str: ["journalctl", "-u", service, "-n", str(lines), "--no-pager"], capture_output=True, text=True, - timeout=30 + timeout=30, ) return result.stdout except Exception: return "" - def get_status(self, service: str) -> Optional[ServiceStatus]: + def get_status(self, service: str) -> ServiceStatus | None: """ Get the status of a systemd service. @@ -237,7 +241,7 @@ def get_status(self, service: str) -> Optional[ServiceStatus]: return status - def explain_status(self, service: str) -> Tuple[bool, str]: + def explain_status(self, service: str) -> tuple[bool, str]: """ Explain a service's status in plain English. @@ -252,15 +256,17 @@ def explain_status(self, service: str) -> Tuple[bool, str]: return False, f"Service '{service}' is not installed on this system." if status.load_state == "masked": - return True, f"Service '{service}' is MASKED (disabled by administrator and cannot be started)." + return ( + True, + f"Service '{service}' is MASKED (disabled by administrator and cannot be started).", + ) # Build explanation parts = [] # Main state state_explanation = SERVICE_STATE_EXPLANATIONS.get( - status.active_state, - f"in an unknown state ({status.active_state})" + status.active_state, f"in an unknown state ({status.active_state})" ) parts.append(f"**{service}** is **{status.active_state}**: {state_explanation}") @@ -294,7 +300,7 @@ def explain_status(self, service: str) -> Tuple[bool, str]: return True, "\n".join(parts) - def diagnose_failure(self, service: str) -> Tuple[bool, str, List[str]]: + def diagnose_failure(self, service: str) -> tuple[bool, str, list[str]]: """ Diagnose why a service failed. @@ -328,7 +334,9 @@ def diagnose_failure(self, service: str) -> Tuple[bool, str, List[str]]: # Analyze logs for common issues log_text = logs.lower() if "permission denied" in log_text: - recommendations.append("- **Permission issue detected**: Check file permissions and service user") + recommendations.append( + "- **Permission issue detected**: Check file permissions and service user" + ) if "address already in use" in log_text: recommendations.append("- **Port conflict**: Another process is using the same port") recommendations.append(" Run: `ss -tlnp | grep ` to find conflicting process") @@ -345,14 +353,14 @@ def diagnose_failure(self, service: str) -> Tuple[bool, str, List[str]]: return len(recommendations) > 0, "\n".join(explanation_parts), log_lines - def get_dependencies(self, service: str) -> Dict[str, List[str]]: + def get_dependencies(self, service: str) -> dict[str, list[str]]: """ Get service dependencies. Returns: Dict with 'wants', 'requires', 'after', 'before' lists """ - deps: Dict[str, List[str]] = { + deps: dict[str, list[str]] = { "wants": [], "requires": [], "after": [], @@ -365,9 +373,9 @@ def get_dependencies(self, service: str) -> Dict[str, List[str]]: service = f"{service}.service" # Get dependency info - returncode, stdout, _ = self._run_systemctl("show", service, - "-p", "Wants,Requires,After,Before,WantedBy,RequiredBy", - "--no-pager") + returncode, stdout, _ = self._run_systemctl( + "show", service, "-p", "Wants,Requires,After,Before,WantedBy,RequiredBy", "--no-pager" + ) if returncode == 0: for line in stdout.split("\n"): @@ -470,10 +478,10 @@ def create_unit_from_description( self, description: str, command: str, - name: Optional[str] = None, - user: Optional[str] = None, - working_dir: Optional[str] = None, - ) -> Tuple[str, str]: + name: str | None = None, + user: str | None = None, + working_dir: str | None = None, + ) -> tuple[str, str]: """ Create a unit file from a simple description. @@ -489,8 +497,8 @@ def create_unit_from_description( """ # Auto-generate name from description if not provided if not name: - name = re.sub(r'[^a-z0-9]+', '-', description.lower())[:40] - name = name.strip('-') + name = re.sub(r"[^a-z0-9]+", "-", description.lower())[:40] + name = name.strip("-") # Detect service type service_type = ServiceType.SIMPLE @@ -562,12 +570,14 @@ def display_status(self, service: str): console.print() success, explanation = self.explain_status(service) if success: - console.print(Panel( - explanation, - title="[bold cyan]Plain English Explanation[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + explanation, + title="[bold cyan]Plain English Explanation[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) def display_diagnosis(self, service: str): """Display failure diagnosis for a service.""" @@ -576,12 +586,14 @@ def display_diagnosis(self, service: str): found_issues, explanation, logs = self.diagnose_failure(service) if explanation: - console.print(Panel( - explanation, - title="[bold yellow]Diagnosis[/bold yellow]", - border_style="yellow", - padding=(1, 2), - )) + console.print( + Panel( + explanation, + title="[bold yellow]Diagnosis[/bold yellow]", + border_style="yellow", + padding=(1, 2), + ) + ) if logs: console.print() @@ -595,11 +607,7 @@ def display_diagnosis(self, service: str): console.print(f"[dim]{line}[/dim]") -def run_systemd_helper( - service: str, - action: str = "status", - verbose: bool = False -) -> int: +def run_systemd_helper(service: str, action: str = "status", verbose: bool = False) -> int: """ Main entry point for cortex systemd command. diff --git a/cortex/update_checker.py b/cortex/update_checker.py index d268a9273..31298e964 100644 --- a/cortex/update_checker.py +++ b/cortex/update_checker.py @@ -16,12 +16,7 @@ import requests -from cortex.version_manager import ( - SemanticVersion, - UpdateChannel, - get_current_version, - is_newer, -) +from cortex.version_manager import SemanticVersion, UpdateChannel, get_current_version, is_newer logger = logging.getLogger(__name__) @@ -46,7 +41,7 @@ class ReleaseInfo: body: str # Release notes (markdown) published_at: str html_url: str - download_url: Optional[str] = None + download_url: str | None = None assets: list[dict] = field(default_factory=list) @classmethod @@ -104,10 +99,10 @@ class UpdateCheckResult: update_available: bool current_version: SemanticVersion - latest_version: Optional[SemanticVersion] = None - latest_release: Optional[ReleaseInfo] = None - error: Optional[str] = None - checked_at: Optional[str] = None + latest_version: SemanticVersion | None = None + latest_release: ReleaseInfo | None = None + error: str | None = None + checked_at: str | None = None from_cache: bool = False @@ -129,7 +124,7 @@ def _ensure_cache_dir(self) -> None: """Ensure cache directory exists.""" CACHE_DIR.mkdir(parents=True, exist_ok=True) - def _get_cached_result(self) -> Optional[UpdateCheckResult]: + def _get_cached_result(self) -> UpdateCheckResult | None: """Get cached update check result if valid.""" if not self.cache_enabled or not UPDATE_CACHE_FILE.exists(): return None @@ -228,8 +223,8 @@ def check(self, force: bool = False) -> UpdateCheckResult: if cached: # Update current version in case we've upgraded cached.current_version = current - cached.update_available = ( - cached.latest_version is not None and is_newer(cached.latest_version, current) + cached.update_available = cached.latest_version is not None and is_newer( + cached.latest_version, current ) return cached @@ -327,7 +322,11 @@ def _filter_by_channel(self, releases: list[ReleaseInfo]) -> list[ReleaseInfo]: if self.channel == UpdateChannel.BETA: # Stable + beta releases - return [r for r in releases if r.version.channel in (UpdateChannel.STABLE, UpdateChannel.BETA)] + return [ + r + for r in releases + if r.version.channel in (UpdateChannel.STABLE, UpdateChannel.BETA) + ] # DEV channel - all releases return releases @@ -368,7 +367,7 @@ def check_for_updates( return checker.check(force=force) -def should_notify_update() -> Optional[ReleaseInfo]: +def should_notify_update() -> ReleaseInfo | None: """Check if we should notify user about an update. This is called on CLI startup. Uses cache to avoid diff --git a/cortex/updater.py b/cortex/updater.py index bddefa614..5c65a3143 100644 --- a/cortex/updater.py +++ b/cortex/updater.py @@ -13,11 +13,12 @@ import subprocess import sys import tempfile +from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from enum import Enum from pathlib import Path -from typing import Callable, Optional +from typing import Optional import requests @@ -58,10 +59,10 @@ class UpdateResult: success: bool status: UpdateStatus previous_version: str - new_version: Optional[str] = None - error: Optional[str] = None - backup_path: Optional[Path] = None - duration_seconds: Optional[float] = None + new_version: str | None = None + error: str | None = None + backup_path: Path | None = None + duration_seconds: float | None = None @dataclass @@ -80,7 +81,7 @@ class Updater: def __init__( self, channel: UpdateChannel = UpdateChannel.STABLE, - progress_callback: Optional[Callable[[str, float], None]] = None, + progress_callback: Callable[[str, float], None] | None = None, ): """Initialize the updater. @@ -116,7 +117,7 @@ def check_update_available(self, force: bool = False) -> UpdateCheckResult: def update( self, - target_version: Optional[str] = None, + target_version: str | None = None, dry_run: bool = False, ) -> UpdateResult: """Perform the update. @@ -250,7 +251,7 @@ def update( error=str(e), ) - def _get_specific_release(self, version: str) -> Optional[ReleaseInfo]: + def _get_specific_release(self, version: str) -> ReleaseInfo | None: """Get information about a specific release version.""" from cortex.update_checker import UpdateChecker @@ -464,7 +465,7 @@ def rollback_to_backup(self, backup_path: str | Path) -> UpdateResult: def download_with_progress( url: str, dest_path: Path, - progress_callback: Optional[Callable[[int, int], None]] = None, + progress_callback: Callable[[int, int], None] | None = None, chunk_size: int = 8192, ) -> bool: """Download a file with progress reporting. diff --git a/cortex/version_manager.py b/cortex/version_manager.py index 990a34726..294ee8a77 100644 --- a/cortex/version_manager.py +++ b/cortex/version_manager.py @@ -11,10 +11,10 @@ from functools import total_ordering from typing import Optional - # Single source of truth for version __version__ = "0.1.0" + # Update channels class UpdateChannel(Enum): STABLE = "stable" @@ -30,8 +30,8 @@ class SemanticVersion: major: int minor: int patch: int - prerelease: Optional[str] = None - build: Optional[str] = None + prerelease: str | None = None + build: str | None = None @classmethod def parse(cls, version_str: str) -> "SemanticVersion": diff --git a/cortex/wifi_driver.py b/cortex/wifi_driver.py index 1f21a338e..c71480cf3 100644 --- a/cortex/wifi_driver.py +++ b/cortex/wifi_driver.py @@ -190,9 +190,7 @@ def __init__(self, verbose: bool = False): self.verbose = verbose self.devices: list[WirelessDevice] = [] - def _run_command( - self, cmd: list[str], timeout: int = 30 - ) -> tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return exit code, stdout, stderr.""" try: result = subprocess.run( @@ -252,12 +250,8 @@ def detect_pci_devices(self) -> list[WirelessDevice]: driver = "" pci_addr = line.split()[0] if line.split() else "" if pci_addr: - _, drv_out, _ = self._run_command( - ["lspci", "-k", "-s", pci_addr] - ) - drv_match = re.search( - r"Kernel driver in use:\s*(\S+)", drv_out - ) + _, drv_out, _ = self._run_command(["lspci", "-k", "-s", pci_addr]) + drv_match = re.search(r"Kernel driver in use:\s*(\S+)", drv_out) if drv_match: driver = drv_match.group(1) @@ -327,7 +321,7 @@ def detect_all_devices(self) -> list[WirelessDevice]: self.devices.extend(self.detect_usb_devices()) return self.devices - def find_driver(self, device: WirelessDevice) -> Optional[DriverInfo]: + def find_driver(self, device: WirelessDevice) -> DriverInfo | None: """Find appropriate driver for a device.""" # Check by vendor:device ID for driver_name, driver in DRIVER_DATABASE.items(): @@ -447,12 +441,20 @@ def display_status(self): conn_table.add_column("Item", style="cyan") conn_table.add_column("Value") - wifi_status = "[green]Connected[/green]" if connectivity["wifi_connected"] else "[red]Not connected[/red]" + wifi_status = ( + "[green]Connected[/green]" + if connectivity["wifi_connected"] + else "[red]Not connected[/red]" + ) if connectivity["wifi_ssid"]: wifi_status += f" ({connectivity['wifi_ssid']})" conn_table.add_row("WiFi", wifi_status) - bt_status = "[green]Available[/green]" if connectivity["bluetooth_available"] else "[red]Not available[/red]" + bt_status = ( + "[green]Available[/green]" + if connectivity["bluetooth_available"] + else "[red]Not available[/red]" + ) if connectivity["bluetooth_powered"]: bt_status += " (Powered)" conn_table.add_row("Bluetooth", bt_status) @@ -597,7 +599,9 @@ def run_wifi_driver( console.print(f"WiFi: {'Connected' if status['wifi_connected'] else 'Not connected'}") if status["wifi_ssid"]: console.print(f" SSID: {status['wifi_ssid']}") - console.print(f"Bluetooth: {'Available' if status['bluetooth_available'] else 'Not available'}") + console.print( + f"Bluetooth: {'Available' if status['bluetooth_available'] else 'Not available'}" + ) return 0 else: diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py index c6bda82fd..23bc2bbc0 100644 --- a/tests/test_benchmark.py +++ b/tests/test_benchmark.py @@ -8,15 +8,15 @@ import os import tempfile from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.benchmark import ( - BenchmarkResult, + MODEL_REQUIREMENTS, BenchmarkReport, + BenchmarkResult, CortexBenchmark, - MODEL_REQUIREMENTS, run_benchmark, ) @@ -27,11 +27,7 @@ class TestBenchmarkResult: def test_result_creation(self): """Test creating a benchmark result.""" result = BenchmarkResult( - name="Test", - score=75, - raw_value=100.5, - unit="ms", - description="Test benchmark" + name="Test", score=75, raw_value=100.5, unit="ms", description="Test benchmark" ) assert result.name == "Test" assert result.score == 75 @@ -40,12 +36,7 @@ def test_result_creation(self): def test_result_default_description(self): """Test default description is empty.""" - result = BenchmarkResult( - name="Test", - score=50, - raw_value=10.0, - unit="s" - ) + result = BenchmarkResult(name="Test", score=50, raw_value=10.0, unit="s") assert result.description == "" @@ -64,11 +55,7 @@ def test_report_defaults(self): def test_report_to_dict(self): """Test report serialization.""" - report = BenchmarkReport( - timestamp="2025-01-01T00:00:00", - overall_score=75, - rating="Good" - ) + report = BenchmarkReport(timestamp="2025-01-01T00:00:00", overall_score=75, rating="Good") result = report.to_dict() assert result["timestamp"] == "2025-01-01T00:00:00" assert result["overall_score"] == 75 @@ -250,9 +237,7 @@ def test_save_to_history(self, benchmark): benchmark.HISTORY_FILE = Path(tmpdir) / "benchmark_history.json" report = BenchmarkReport( - timestamp="2025-01-01T00:00:00", - overall_score=75, - rating="Good" + timestamp="2025-01-01T00:00:00", overall_score=75, rating="Good" ) benchmark._save_to_history(report) @@ -299,19 +284,13 @@ def test_detect_nvidia_gpu_not_available(self, benchmark): def test_detect_nvidia_gpu_available(self, benchmark): """Test when NVIDIA GPU is detected.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="NVIDIA GeForce RTX 3080" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="NVIDIA GeForce RTX 3080") assert benchmark._detect_nvidia_gpu() is True def test_get_nvidia_vram(self, benchmark): """Test getting NVIDIA VRAM.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="10240" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="10240") assert benchmark._get_nvidia_vram() == 10240 diff --git a/tests/test_gpu_manager.py b/tests/test_gpu_manager.py index ce50c6699..be7440b6e 100644 --- a/tests/test_gpu_manager.py +++ b/tests/test_gpu_manager.py @@ -4,18 +4,18 @@ Issue: #454 - Hybrid GPU (Optimus) Manager """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.gpu_manager import ( + APP_GPU_RECOMMENDATIONS, + BATTERY_IMPACT, GPUDevice, GPUMode, GPUState, GPUVendor, HybridGPUManager, - BATTERY_IMPACT, - APP_GPU_RECOMMENDATIONS, run_gpu_manager, ) @@ -47,10 +47,7 @@ class TestGPUDevice: def test_default_values(self): """Test default device values.""" - device = GPUDevice( - vendor=GPUVendor.INTEL, - name="Intel HD Graphics" - ) + device = GPUDevice(vendor=GPUVendor.INTEL, name="Intel HD Graphics") assert device.vendor == GPUVendor.INTEL assert device.name == "Intel HD Graphics" assert device.driver == "" @@ -167,11 +164,7 @@ def test_run_command_not_found(self, manager): def test_run_command_success(self, manager): """Test successful command execution.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="output", - stderr="" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="output", stderr="") code, stdout, stderr = manager._run_command(["test"]) assert code == 0 assert stdout == "output" @@ -298,9 +291,7 @@ def manager(self): def test_switch_mode_non_hybrid(self, manager): """Test switching on non-hybrid system.""" - state = GPUState(devices=[ - GPUDevice(vendor=GPUVendor.INTEL, name="Intel") - ]) + state = GPUState(devices=[GPUDevice(vendor=GPUVendor.INTEL, name="Intel")]) with patch.object(manager, "get_state") as mock_state: mock_state.return_value = state @@ -310,10 +301,12 @@ def test_switch_mode_non_hybrid(self, manager): def test_switch_mode_with_prime_select(self, manager): """Test switching with prime-select available.""" - state = GPUState(devices=[ - GPUDevice(vendor=GPUVendor.INTEL, name="Intel"), - GPUDevice(vendor=GPUVendor.NVIDIA, name="NVIDIA"), - ]) + state = GPUState( + devices=[ + GPUDevice(vendor=GPUVendor.INTEL, name="Intel"), + GPUDevice(vendor=GPUVendor.NVIDIA, name="NVIDIA"), + ] + ) with patch.object(manager, "get_state") as mock_state: mock_state.return_value = state diff --git a/tests/test_health_score.py b/tests/test_health_score.py index 1db8d0731..890008e76 100644 --- a/tests/test_health_score.py +++ b/tests/test_health_score.py @@ -9,7 +9,7 @@ import tempfile from datetime import datetime from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest diff --git a/tests/test_printer_setup.py b/tests/test_printer_setup.py index ffe5941dc..6e7eded0d 100644 --- a/tests/test_printer_setup.py +++ b/tests/test_printer_setup.py @@ -4,18 +4,18 @@ Issue: #451 - Printer/Scanner Auto-Setup """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.printer_setup import ( + DRIVER_PACKAGES, + SCANNER_PACKAGES, ConnectionType, DeviceType, DriverInfo, PrinterDevice, PrinterSetup, - DRIVER_PACKAGES, - SCANNER_PACKAGES, run_printer_setup, ) @@ -165,7 +165,11 @@ def test_detect_usb_printers_parses_lsusb(self, setup): def test_detect_usb_printers_empty(self, setup): """Test when no printers detected.""" with patch.object(setup, "_run_command") as mock_cmd: - mock_cmd.return_value = (0, "Bus 001 Device 001: ID 1d6b:0002 Linux Foundation 2.0 root hub", "") + mock_cmd.return_value = ( + 0, + "Bus 001 Device 001: ID 1d6b:0002 Linux Foundation 2.0 root hub", + "", + ) devices = setup.detect_usb_printers() assert devices == [] diff --git a/tests/test_stdin_handler.py b/tests/test_stdin_handler.py index 9af2d488f..7f3bd0188 100644 --- a/tests/test_stdin_handler.py +++ b/tests/test_stdin_handler.py @@ -7,7 +7,7 @@ import io import json import sys -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest @@ -188,7 +188,7 @@ def test_detect_json(self): content = '{"key": "value"}' assert detect_content_type(content) == "json" - content = '[1, 2, 3]' + content = "[1, 2, 3]" assert detect_content_type(content) == "json" def test_detect_python_traceback(self): @@ -260,7 +260,7 @@ def test_analyze_git_diff(self): def test_analyze_json_array(self): """Test JSON array analysis.""" data = StdinData( - content='[1, 2, 3, 4, 5]', + content="[1, 2, 3, 4, 5]", line_count=1, byte_count=15, ) @@ -338,9 +338,7 @@ def test_run_info_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -358,9 +356,7 @@ def test_run_unknown_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -380,9 +376,7 @@ def test_run_passthrough_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="hello world", line_count=1, byte_count=11 - ), + return_value=StdinData(content="hello world", line_count=1, byte_count=11), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -402,9 +396,7 @@ def test_run_stats_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -484,7 +476,7 @@ def test_read_error(self): handler = StdinHandler() with patch("sys.stdin.isatty", return_value=False): - with patch("sys.stdin.read", side_effect=IOError("Read error")): + with patch("sys.stdin.read", side_effect=OSError("Read error")): data = handler.read_stdin() assert data.is_empty diff --git a/tests/test_systemd_helper.py b/tests/test_systemd_helper.py index 681a28eda..4ee3491cb 100644 --- a/tests/test_systemd_helper.py +++ b/tests/test_systemd_helper.py @@ -4,18 +4,18 @@ Issue: #448 - Systemd Service Helper (Plain English) """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.systemd_helper import ( + FAILURE_SOLUTIONS, + SERVICE_STATE_EXPLANATIONS, + SUB_STATE_EXPLANATIONS, ServiceConfig, ServiceStatus, ServiceType, SystemdHelper, - SERVICE_STATE_EXPLANATIONS, - SUB_STATE_EXPLANATIONS, - FAILURE_SOLUTIONS, run_systemd_helper, ) @@ -25,11 +25,7 @@ class TestServiceConfig: def test_default_values(self): """Test default configuration values.""" - config = ServiceConfig( - name="test", - description="Test service", - exec_start="/usr/bin/test" - ) + config = ServiceConfig(name="test", description="Test service", exec_start="/usr/bin/test") assert config.name == "test" assert config.service_type == ServiceType.SIMPLE assert config.restart == "on-failure" @@ -137,11 +133,7 @@ def test_run_systemctl_not_found(self, helper): def test_run_systemctl_success(self, helper): """Test successful systemctl command.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="ActiveState=active", - stderr="" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="ActiveState=active", stderr="") code, stdout, stderr = helper._run_systemctl("status", "test") assert code == 0 assert "active" in stdout.lower() @@ -149,6 +141,7 @@ def test_run_systemctl_success(self, helper): def test_run_systemctl_timeout(self, helper): """Test systemctl timeout handling.""" import subprocess + with patch("subprocess.run") as mock_run: mock_run.side_effect = subprocess.TimeoutExpired("cmd", 30) code, stdout, stderr = helper._run_systemctl("status", "test") diff --git a/tests/test_update_checker.py b/tests/test_update_checker.py index b9026a8af..7356e0840 100644 --- a/tests/test_update_checker.py +++ b/tests/test_update_checker.py @@ -13,8 +13,8 @@ from cortex.update_checker import ( CACHE_TTL_SECONDS, ReleaseInfo, - UpdateCheckResult, UpdateChecker, + UpdateCheckResult, check_for_updates, should_notify_update, ) @@ -99,9 +99,7 @@ def setUp(self): """Set up test fixtures.""" # Use temp directory for cache self.temp_dir = tempfile.mkdtemp() - self.cache_patch = patch( - "cortex.update_checker.CACHE_DIR", Path(self.temp_dir) - ) + self.cache_patch = patch("cortex.update_checker.CACHE_DIR", Path(self.temp_dir)) self.cache_patch.start() self.cache_file_patch = patch( "cortex.update_checker.UPDATE_CACHE_FILE", diff --git a/tests/test_updater.py b/tests/test_updater.py index 9219d9c32..1b8b0eb9b 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -14,9 +14,9 @@ from cortex.update_checker import ReleaseInfo, UpdateCheckResult from cortex.updater import ( BackupInfo, + Updater, UpdateResult, UpdateStatus, - Updater, download_with_progress, verify_checksum, ) diff --git a/tests/test_wifi_driver.py b/tests/test_wifi_driver.py index 530ebacf4..c52df2614 100644 --- a/tests/test_wifi_driver.py +++ b/tests/test_wifi_driver.py @@ -4,19 +4,19 @@ Issue: #444 - WiFi/Bluetooth Driver Auto-Matcher """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.wifi_driver import ( + BLUETOOTH_DRIVERS, + DRIVER_DATABASE, ConnectionType, DeviceType, DriverInfo, DriverSource, WirelessDevice, WirelessDriverMatcher, - DRIVER_DATABASE, - BLUETOOTH_DRIVERS, run_wifi_driver, ) From 8c960998f3256a6791f1a3262624019df71a51ee Mon Sep 17 00:00:00 2001 From: Anshgrover23 Date: Thu, 15 Jan 2026 21:29:53 +0530 Subject: [PATCH 2/2] fix lint-2 --- cortex/benchmark.py | 5 ++--- cortex/branding.py | 2 -- cortex/gpu_manager.py | 3 --- cortex/health_score.py | 4 ---- cortex/output_formatter.py | 8 ++------ cortex/printer_setup.py | 5 +---- cortex/systemd_helper.py | 3 --- cortex/updater.py | 9 +-------- tests/test_stdin_handler.py | 3 +-- 9 files changed, 7 insertions(+), 35 deletions(-) diff --git a/cortex/benchmark.py b/cortex/benchmark.py index 8687db0fc..a98125944 100644 --- a/cortex/benchmark.py +++ b/cortex/benchmark.py @@ -15,15 +15,14 @@ from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path -from typing import Any, Optional +from typing import Any from rich import box -from rich.console import Console from rich.panel import Panel from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn from rich.table import Table -from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print +from cortex.branding import console, cx_header, cx_print # Model recommendations based on system capabilities MODEL_REQUIREMENTS = { diff --git a/cortex/branding.py b/cortex/branding.py index f3ed47b32..ea85ab3fd 100644 --- a/cortex/branding.py +++ b/cortex/branding.py @@ -11,8 +11,6 @@ - Consistent visual language """ -from typing import Optional - from rich import box from rich.console import Console from rich.panel import Panel diff --git a/cortex/gpu_manager.py b/cortex/gpu_manager.py index 171708ddf..8b25eeb2d 100644 --- a/cortex/gpu_manager.py +++ b/cortex/gpu_manager.py @@ -7,16 +7,13 @@ Issue: #454 """ -import os import re import subprocess from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Optional from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table diff --git a/cortex/health_score.py b/cortex/health_score.py index 2e68c9bd8..497859f89 100644 --- a/cortex/health_score.py +++ b/cortex/health_score.py @@ -7,15 +7,11 @@ """ import json -import os import subprocess -import time -from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime from enum import Enum from pathlib import Path -from typing import Optional from rich.console import Console from rich.panel import Panel diff --git a/cortex/output_formatter.py b/cortex/output_formatter.py index 73f272856..ea1583ada 100644 --- a/cortex/output_formatter.py +++ b/cortex/output_formatter.py @@ -9,12 +9,11 @@ from collections.abc import Generator from contextlib import contextmanager -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum -from typing import Any from rich import box -from rich.console import Console, Group +from rich.console import Console from rich.live import Live from rich.panel import Panel from rich.progress import ( @@ -28,11 +27,8 @@ TimeElapsedColumn, TimeRemainingColumn, ) -from rich.spinner import Spinner from rich.status import Status -from rich.style import Style from rich.table import Table -from rich.text import Text from rich.tree import Tree console = Console() diff --git a/cortex/printer_setup.py b/cortex/printer_setup.py index 5b1733a67..86f16219a 100644 --- a/cortex/printer_setup.py +++ b/cortex/printer_setup.py @@ -8,13 +8,10 @@ import re import subprocess -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum -from pathlib import Path -from typing import Optional from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table diff --git a/cortex/systemd_helper.py b/cortex/systemd_helper.py index bc63775d0..e29d2b5bf 100644 --- a/cortex/systemd_helper.py +++ b/cortex/systemd_helper.py @@ -7,16 +7,13 @@ Issue: #448 """ -import os import re import subprocess from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Optional from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table from rich.tree import Tree diff --git a/cortex/updater.py b/cortex/updater.py index 5c65a3143..02c39116c 100644 --- a/cortex/updater.py +++ b/cortex/updater.py @@ -12,23 +12,16 @@ import shutil import subprocess import sys -import tempfile from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from enum import Enum from pathlib import Path -from typing import Optional import requests from cortex.update_checker import ReleaseInfo, UpdateCheckResult, check_for_updates -from cortex.version_manager import ( - SemanticVersion, - UpdateChannel, - get_current_version, - get_version_string, -) +from cortex.version_manager import SemanticVersion, UpdateChannel, get_version_string logger = logging.getLogger(__name__) diff --git a/tests/test_stdin_handler.py b/tests/test_stdin_handler.py index 7f3bd0188..51524ea73 100644 --- a/tests/test_stdin_handler.py +++ b/tests/test_stdin_handler.py @@ -4,10 +4,9 @@ Issue: #271 - Stdin Piping Support for Log Analysis """ -import io import json import sys -from unittest.mock import MagicMock, patch +from unittest.mock import patch import pytest