Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions cortex/benchmark.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Check failure on line 18 in cortex/benchmark.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/benchmark.py:18:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 18 in cortex/benchmark.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/benchmark.py:18:1: UP035 `typing.List` is deprecated, use `list` instead

Check failure on line 18 in cortex/benchmark.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/benchmark.py:18:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 18 in cortex/benchmark.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/benchmark.py:18:1: UP035 `typing.List` is deprecated, use `list` instead

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 = {
Expand Down Expand Up @@ -57,11 +57,11 @@

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]:
Expand Down Expand Up @@ -95,7 +95,7 @@

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."""
Expand Down Expand Up @@ -418,7 +418,7 @@
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.

Expand Down Expand Up @@ -465,7 +465,7 @@

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.

Expand Down
18 changes: 9 additions & 9 deletions cortex/branding.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
- Consistent visual language
"""

from typing import List, Optional, Tuple

Check failure on line 14 in cortex/branding.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/branding.py:14:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 14 in cortex/branding.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/branding.py:14:1: UP035 `typing.List` is deprecated, use `list` instead

Check failure on line 14 in cortex/branding.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/branding.py:14:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 14 in cortex/branding.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/branding.py:14:1: UP035 `typing.List` is deprecated, use `list` instead

from rich import box
from rich.console import Console
Expand Down Expand Up @@ -141,8 +141,8 @@

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:
"""
Expand Down Expand Up @@ -175,7 +175,7 @@

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.
Expand Down Expand Up @@ -221,10 +221,10 @@


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.
Expand Down Expand Up @@ -255,7 +255,7 @@


def cx_package_table(
packages: List[Tuple[str, str, str]],
packages: list[tuple[str, str, str]],
title: str = "Packages",
) -> None:
"""
Expand Down Expand Up @@ -293,7 +293,7 @@
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.

Expand Down
12 changes: 6 additions & 6 deletions cortex/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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"
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
24 changes: 12 additions & 12 deletions cortex/gpu_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.List` is deprecated, use `list` instead

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / Lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.Dict` is deprecated, use `dict` instead

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.Tuple` is deprecated, use `tuple` instead

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.List` is deprecated, use `list` instead

Check failure on line 16 in cortex/gpu_manager.py

View workflow job for this annotation

GitHub Actions / lint

Ruff (UP035)

cortex/gpu_manager.py:16:1: UP035 `typing.Dict` is deprecated, use `dict` instead

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):
Expand Down Expand Up @@ -61,8 +61,8 @@
"""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 = ""
Expand All @@ -83,7 +83,7 @@
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)
Expand Down Expand Up @@ -126,9 +126,9 @@

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(
Expand All @@ -143,7 +143,7 @@
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 = []

Expand Down Expand Up @@ -172,7 +172,7 @@

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()

Expand Down Expand Up @@ -200,7 +200,7 @@
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",
Expand Down Expand Up @@ -292,7 +292,7 @@
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.

Expand Down Expand Up @@ -385,7 +385,7 @@
# 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"})

Expand Down Expand Up @@ -519,7 +519,7 @@

def run_gpu_manager(
action: str = "status",
mode: Optional[str] = None,
mode: str | None = None,
verbose: bool = False
) -> int:
"""
Expand Down
9 changes: 5 additions & 4 deletions cortex/health_score.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = {
Expand All @@ -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]:
Expand All @@ -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):
Expand Down
16 changes: 8 additions & 8 deletions cortex/licensing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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']}")
Expand All @@ -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


Expand Down
Loading
Loading