|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import re |
3 | 4 | import json as json_lib |
| 5 | +from typing import Any, Dict, List |
4 | 6 |
|
5 | 7 | import click |
| 8 | +from tabulate import tabulate |
6 | 9 |
|
7 | 10 | from together import Together, omit |
| 11 | +from together.types import HardwareListResponse |
8 | 12 | from together.lib.cli.api._utils import handle_api_errors |
9 | 13 | from together.lib.utils.serializer import datetime_serializer |
10 | 14 |
|
|
21 | 25 | @handle_api_errors("Endpoints") |
22 | 26 | def hardware(client: Together, model: str | None, json: bool, available: bool) -> None: |
23 | 27 | """List all available hardware options, optionally filtered by model.""" |
24 | | - message = "Available hardware options:" if available else "All hardware options:" |
25 | | - click.echo(message, err=True) |
26 | 28 | hardware_options = client.hardware.list(model=model or omit) |
27 | | - # hardware_options = client.endpoints.list_hardware(model) |
| 29 | + |
28 | 30 | if available: |
29 | 31 | hardware_options.data = [ |
30 | 32 | hardware |
31 | 33 | for hardware in hardware_options.data |
32 | 34 | if hardware.availability is not None and hardware.availability.status == "available" |
33 | 35 | ] |
34 | | - |
35 | 36 | if json: |
36 | 37 | json_output = [hardware.model_dump() for hardware in hardware_options.data] |
37 | 38 | click.echo(json_lib.dumps(json_output, default=datetime_serializer, indent=2)) |
38 | 39 | else: |
39 | | - for hardware in hardware_options.data: |
40 | | - click.echo(f" {hardware.id}", err=True) |
| 40 | + _format_hardware_options(hardware_options, show_availability=model is not None) |
| 41 | + |
| 42 | + |
| 43 | +def _format_hardware_options(hardware_options: HardwareListResponse, show_availability: bool = True) -> None: |
| 44 | + display_list: List[Dict[str, Any]] = [] |
| 45 | + |
| 46 | + for hw in hardware_options.data: |
| 47 | + data = { |
| 48 | + "Hardware ID": hw.id, |
| 49 | + "GPU": re.sub(r"\-\d+[a-zA-Z][a-zA-Z]$", "", hw.specs.gpu_type) |
| 50 | + if hw.specs and hw.specs.gpu_type |
| 51 | + else "N/A", |
| 52 | + "Memory": f"{int(hw.specs.gpu_memory)}GB" if hw.specs else "N/A", |
| 53 | + "Count": hw.specs.gpu_count if hw.specs else "N/A", |
| 54 | + "Price (per minute)": (f"${hw.pricing.cents_per_minute / 100:.2f}" if hw.pricing else "N/A"), |
| 55 | + } |
| 56 | + |
| 57 | + if show_availability: |
| 58 | + status_display = "—" |
| 59 | + if hw.availability: |
| 60 | + status = hw.availability.status |
| 61 | + # Add visual indicators for status |
| 62 | + if status == "available": |
| 63 | + status_display = click.style("✓ available", fg="green") |
| 64 | + elif status == "unavailable": |
| 65 | + status_display = click.style("✗ unavailable", fg="red") |
| 66 | + else: # insufficient |
| 67 | + status_display = click.style("⚠ insufficient", fg="yellow") |
| 68 | + data["availability"] = status_display |
| 69 | + display_list.append(data) |
| 70 | + |
| 71 | + click.echo(tabulate(display_list, headers="keys", numalign="left")) |
0 commit comments