diff --git a/cortex/benchmark.py b/cortex/benchmark.py index 1b35b45b..92dc0382 100644 --- a/cortex/benchmark.py +++ b/cortex/benchmark.py @@ -20,10 +20,10 @@ 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.""" @@ -418,7 +418,7 @@ def _benchmark_token_generation(self) -> BenchmarkResult: 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 +465,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. diff --git a/cortex/branding.py b/cortex/branding.py index de412be3..84e3972c 100644 --- a/cortex/branding.py +++ b/cortex/branding.py @@ -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. diff --git a/cortex/cli.py b/cortex/cli.py index c13d5db8..d68d15c9 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, @@ -37,7 +34,10 @@ ServiceStatus, UninstallImpactAnalyzer, ) +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 # CLI Help Constants HELP_SKIP_CONFIRM = "Skip confirmation prompt" @@ -1604,11 +1604,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: @@ -2932,7 +2932,7 @@ def main(): f"[green]{update_release.version}[/green]" ) console.print( - f" [dim]Run 'cortex update' to upgrade[/dim]" + " [dim]Run 'cortex update' to upgrade[/dim]" ) console.print() except Exception: diff --git a/cortex/gpu_manager.py b/cortex/gpu_manager.py index ee6245fd..5135fb8c 100644 --- a/cortex/gpu_manager.py +++ b/cortex/gpu_manager.py @@ -20,7 +20,7 @@ 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,9 +126,9 @@ 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( @@ -143,7 +143,7 @@ def _run_command(self, cmd: List[str], timeout: int = 10) -> Tuple[int, str, str 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 +172,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,7 +200,7 @@ 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", @@ -292,7 +292,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. @@ -385,7 +385,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"}) @@ -519,7 +519,7 @@ def display_app_recommendations(self): def run_gpu_manager( action: str = "status", - mode: Optional[str] = None, + mode: str | None = None, verbose: bool = False ) -> int: """ diff --git a/cortex/health_score.py b/cortex/health_score.py index de12e7c2..8344e6aa 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 @@ -477,7 +478,7 @@ 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 = { @@ -497,7 +498,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 +509,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): diff --git a/cortex/licensing.py b/cortex/licensing.py index 9fef8173..b20f8616 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 @@ -81,9 +81,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 +107,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: @@ -251,7 +251,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: @@ -292,7 +292,7 @@ def activate_license(license_key: str) -> bool: # 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 +303,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 diff --git a/cortex/output_formatter.py b/cortex/output_formatter.py index e4a868be..476b11e1 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, List, Optional, Tuple 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 68caafe0..e405db98 100644 --- a/cortex/printer_setup.py +++ b/cortex/printer_setup.py @@ -18,7 +18,7 @@ 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,7 +103,7 @@ 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( @@ -128,7 +128,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 = [] @@ -171,7 +171,7 @@ def detect_usb_printers(self) -> List[PrinterDevice]: 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 = [] @@ -198,7 +198,7 @@ def detect_network_printers(self) -> List[PrinterDevice]: return devices - def detect_configured_printers(self) -> List[PrinterDevice]: + def detect_configured_printers(self) -> list[PrinterDevice]: """Get list of already configured printers.""" devices = [] @@ -234,7 +234,7 @@ def detect_configured_printers(self) -> List[PrinterDevice]: return devices - def detect_scanners(self) -> List[PrinterDevice]: + def detect_scanners(self) -> list[PrinterDevice]: """Detect scanners using SANE.""" devices = [] @@ -288,7 +288,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 +319,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 +336,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. @@ -395,7 +395,7 @@ 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" @@ -411,7 +411,7 @@ def test_print(self, printer_name: str) -> Tuple[bool, str]: 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" diff --git a/cortex/semver_resolver.py b/cortex/semver_resolver.py index 38ff30b1..27a51ca8 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.""" @@ -184,7 +184,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: @@ -261,7 +261,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 +283,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: @@ -514,7 +514,7 @@ def suggest_resolutions( def _find_common_version_strategy( self, conflict: VersionConflict - ) -> Optional[ResolutionStrategy]: + ) -> ResolutionStrategy | None: """Try to find a common version that satisfies all constraints.""" constraints = [d.constraint for d in conflict.dependencies] @@ -615,7 +615,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. diff --git a/cortex/stdin_handler.py b/cortex/stdin_handler.py index 8c5ce4dd..d9e57103 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) @@ -309,7 +309,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 4a125e33..e837ddcb 100644 --- a/cortex/systemd_helper.py +++ b/cortex/systemd_helper.py @@ -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 = { @@ -99,17 +99,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 +130,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,7 +150,7 @@ 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: @@ -179,7 +179,7 @@ def _run_journalctl(self, service: str, lines: int = 50) -> str: 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 +237,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. @@ -294,7 +294,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. @@ -345,14 +345,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": [], @@ -470,10 +470,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. diff --git a/cortex/update_checker.py b/cortex/update_checker.py index d268a927..32c64e1a 100644 --- a/cortex/update_checker.py +++ b/cortex/update_checker.py @@ -46,7 +46,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 +104,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 +129,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 @@ -368,7 +368,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 bddefa61..5c65a314 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 990a3472..676c5b2e 100644 --- a/cortex/version_manager.py +++ b/cortex/version_manager.py @@ -11,7 +11,6 @@ from functools import total_ordering from typing import Optional - # Single source of truth for version __version__ = "0.1.0" @@ -30,8 +29,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 1f21a338..0013e42d 100644 --- a/cortex/wifi_driver.py +++ b/cortex/wifi_driver.py @@ -327,7 +327,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():