diff --git a/.github/workflows/dependabot-uv-lock-commit.yml b/.github/workflows/dependabot-uv-lock-commit.yml new file mode 100644 index 0000000..7d2de5f --- /dev/null +++ b/.github/workflows/dependabot-uv-lock-commit.yml @@ -0,0 +1,50 @@ +name: "Dependabot: Commit uv.lock" +# Privileged companion to dependabot-uv-lock.yml. +# Triggered only after the unprivileged lock workflow succeeds. Downloads the +# artifact produced there and commits it to the PR branch. No untrusted code +# from the PR is executed here — only git operations on a known-good artifact. +permissions: + contents: read +on: + workflow_run: + workflows: ["Dependabot: Update uv.lock"] + types: [completed] +jobs: + commit: + runs-on: ubuntu-latest + if: > + github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.actor.login == 'dependabot[bot]' + + permissions: + contents: write + steps: + # Checkout by exact SHA so we commit on top of what the lock job saw. + - name: Checkout PR branch at workflow HEAD SHA + uses: actions/checkout@v6 + with: + ref: ${{ github.event.workflow_run.head_sha }} + token: ${{ secrets.DEPENDABOT_PAT }} + # Download after checkout so the artifact lands directly in the workspace, + # overwriting the existing uv.lock with the freshly-generated one. + # Push target is passed via env to avoid shell-injection from branch names. + - name: Download uv.lock artifact + uses: actions/download-artifact@v7 + with: + name: uv-lock + run-id: ${{ github.event.workflow_run.id }} + github-token: ${{ secrets.GITHUB_TOKEN }} + - name: Commit and push changes + env: + HEAD_BRANCH: ${{ github.event.workflow_run.head_branch }} + run: | + git config --global user.name "github-actions[bot]" + git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com" + + git add uv.lock + + if ! git diff --staged --quiet; then + git commit -m "chore(dependabot): synchronize uv.lock" + git push origin "HEAD:${HEAD_BRANCH}" + else + echo "uv.lock is already up to date." + fi diff --git a/.github/workflows/dependabot-uv-lock.yml b/.github/workflows/dependabot-uv-lock.yml index 75df779..1686568 100644 --- a/.github/workflows/dependabot-uv-lock.yml +++ b/.github/workflows/dependabot-uv-lock.yml @@ -1,8 +1,11 @@ name: "Dependabot: Update uv.lock" +# Runs in the unprivileged pull_request context — no secrets, no write access. +# The generated uv.lock is uploaded as an artifact for the companion +# dependabot-uv-lock-commit workflow to consume once this run completes. permissions: contents: read on: - pull_request_target: + pull_request: types: [opened, synchronize] jobs: lock: @@ -24,32 +27,3 @@ jobs: with: name: uv-lock path: uv.lock - commit: - needs: lock - runs-on: ubuntu-latest - if: github.actor == 'dependabot[bot]' - permissions: - contents: write - steps: - - name: Checkout the PR branch - uses: actions/checkout@v6 - with: - ref: ${{ github.head_ref }} - token: ${{ secrets.DEPENDABOT_PAT }} - - name: Download uv.lock - uses: actions/download-artifact@v7 - with: - name: uv-lock - - name: Commit and push changes - run: | - git config --global user.name "github-actions[bot]" - git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com" - - git add uv.lock - - if ! git diff --staged --quiet; then - git commit -m "chore(dependabot): synchronize uv.lock" - git push - else - echo "uv.lock is already up to date." - fi diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 50e5432..9b43fc0 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -28,6 +28,6 @@ repos: - id: yamlfmt exclude: ^mkdocs\.yml$ - repo: https://github.com/rhysd/actionlint - rev: v1.7.10 + rev: v1.7.11 hooks: - id: actionlint diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..cc2d69c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,69 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to +[Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [1.0.0] - 2026-02-23 + +### Breaking Changes + +- Removed deprecated shim methods from `PrusaConnectClient` that were present in + the alpha releases. The service-based API is now the sole public interface: + + | Removed method | Replacement | + | -------------------------------- | ----------------------------------------- | + | `client.get_printers()` | `client.printers.list_printers()` | + | `client.get_printer(uuid)` | `client.printers.get(uuid)` | + | `client.get_cameras()` | `client.cameras.list()` | + | `client.get_teams()` | `client.teams.list_teams()` | + | `client.get_team(id)` | `client.teams.get(id)` | + | `client.send_command(uuid, cmd)` | `client.printers.send_command(uuid, cmd)` | + +### Added + +- **Service-based API:** Resources are now accessed through dedicated service + objects on the client — `client.printers`, `client.cameras`, `client.teams`, + `client.files`, `client.jobs`, and `client.stats`. +- **`prusactl` CLI** with full subcommand coverage: `printer`, `camera`, `team`, + `job`, `file`, `stats`, and `auth`. +- **Statistics service** (`client.stats`) for per-printer material usage, print + time, planned tasks, and job success metrics. +- **Printer command discovery** — `client.printers.get_supported_commands(uuid)` + with optional disk caching and TTL. +- **Validated command execution** — `client.execute_printer_command()` validates + arguments against the printer's reported command schema before sending. +- **Camera Signal.IO signaling client** (`PrusaCameraClient`) for pan/tilt + control and image adjustment via the Prusa signaling protocol. +- **G-code metadata parser** (`client.validate_gcode(path)`) for pre-flight + checks before uploading. +- **Persistent credential caching** — CLI credentials are stored in the platform + config directory and auto-loaded by the SDK. +- **CLI output format control** — the `--format` global flag selects between + `rich` (coloured tables; default when stdout is a TTY), `plain` + (tab-separated text with no ANSI escapes or table borders; default when + stdout is not a TTY), and `json` (JSON array per table to stdout, all status + messages to stderr). +- The active format can also be set via the `output_format` key in `config.json` + or the `PRUSACTL_OUTPUT_FORMAT` environment variable; the priority is: CLI + flag → env var → config file → TTY auto-detect. + +### Changed + +- CLI rewritten with [Cyclopts](https://github.com/BrianPugh/cyclopts) for + richer help output and `--verbose` / `--debug` global flags. +- Pydantic models split into focused submodules under + `prusa.connect.client.models`. +- `AppConfig` is now fetched at client init time to validate that the server + supports the `PRUSA_AUTH` backend. + +## [1.0.0a2] - 2025-01-13 + +### Changed + +- Refactored CLI to use Cyclopts; split monolithic models into service modules; + added stats commands and documentation site. + +## [1.0.0a0] - Initial alpha release diff --git a/README.md b/README.md index 090519a..15b6eb6 100644 --- a/README.md +++ b/README.md @@ -11,8 +11,8 @@ provides a frictionless, strongly-typed interface for the Prusa Connect API. > > This SDK is not an officially supported or endorsed product of Prusa Research. > It is developed and maintained by an independent developer and is not -> affiliated with Prusa Research. See [Motivation & Design](#motivation-design) -> for more information. +> affiliated with Prusa Research. See +> [Motivation & Design](#motivation-and-design) for more information. **Features:** @@ -20,7 +20,7 @@ provides a frictionless, strongly-typed interface for the Prusa Connect API. - **Strong Typing:** Full Pydantic models for printers, jobs, cameras, and files. - **Batteries Included:** Retries, timeouts, and error handling out of the box. -- **CLI Tool:** Managing printers from the terminal. +- **CLI Tool:** Manage printers from the terminal with `prusactl`. ## Installation @@ -36,7 +36,31 @@ Or install the lightweight library only: pip install prusa-connect-sdk-client ``` -## Motivation & Design +## Quick Start + +```python +from prusa.connect.client import PrusaConnectClient + +# Credentials are automatically loaded from the CLI session +# (run `prusactl auth login` first) +client = PrusaConnectClient() + +for printer in client.printers.list_printers(): + status = printer.printer_state or "UNKNOWN" + print(f"- {printer.name} ({status})") +``` + +Resources are grouped by service — `client.printers`, `client.cameras`, +`client.teams`, `client.files`, `client.jobs`, and `client.stats`. + +## Documentation + +Full documentation including the CLI reference, SDK quickstart, and API +reference is available at: + +**** + +## Motivation and Design My motivation to create this library is to provide a frictionless, strongly-typed interface for the Prusa Connect API. I want to be able to monitor @@ -56,3 +80,13 @@ gladly accept Prusameters towards a new Core-generation printer. 😉 My Printables Profile: :simple-printables: [dcode](https://www.printables.com/@dcode_3006269) + +## Contributing + +Contributions are welcome! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for +development setup, testing, and pull request guidelines. + +## License + +This project is licensed under the +[GNU Affero General Public License v3.0 or later](LICENSE). diff --git a/docs/contributing.md b/docs/CONTRIBUTING.md similarity index 100% rename from docs/contributing.md rename to docs/CONTRIBUTING.md diff --git a/docs/LICENSE b/docs/LICENSE new file mode 120000 index 0000000..ea5b606 --- /dev/null +++ b/docs/LICENSE @@ -0,0 +1 @@ +../LICENSE \ No newline at end of file diff --git a/docs/cli/quickstart.md b/docs/cli/quickstart.md index f21b34e..5f1a583 100644 --- a/docs/cli/quickstart.md +++ b/docs/cli/quickstart.md @@ -111,6 +111,23 @@ prusactl printer --help prusactl stats --help ``` +## Step 10: Change Output Format + +By default, `prusactl` uses `rich` output (colored tables and text) when running +in a terminal, and `plain` output (tab-separated) when piped or redirected. You +can explicitly set the output format using the `--format` flag: + +```bash +# JSON output (useful for jq or other scripts) +prusactl printer list --format json + +# Plain text (tab-separated) +prusactl printer list --format plain + +# Force rich output even when redirected +prusactl printer list --format rich +``` + ## Configuration File Settings like default printer, team, and camera IDs are stored in a JSON file in @@ -128,8 +145,13 @@ You can edit this file directly. Supported keys: { "default_printer_id": "your-printer-uuid", "default_team_id": 12345, - "default_camera_id": "your-camera-id" + "default_camera_id": "your-camera-id", + "output_format": "json" } ``` -Environment variables (e.g. `DEFAULT_PRINTER_ID`) override file values. +Environment variables override file values: + +- `PRUSACTL_OUTPUT_FORMAT`: Set to `rich`, `plain`, or `json`. +- `DEFAULT_PRINTER_ID`: Override the default printer UUID. +- `DEFAULT_TEAM_ID`: Override the default team ID. diff --git a/docs/examples.md b/docs/examples.md index d2b5cb2..0b476b6 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -16,11 +16,11 @@ from prusa.connect.client import PrusaConnectClient client = PrusaConnectClient() -# Get your printer's UUID (e.g., from client.get_printers()) +# Get your printer's UUID (e.g., from client.printers.list_printers()) printer_uuid = "c0ffee-uuid-1234" # Pause the print -client.send_command(printer_uuid, "PAUSE_PRINT") +client.pause_print(printer_uuid) print("Printer paused.") ``` @@ -32,7 +32,7 @@ Fetch the latest snapshot from your printer's camera. from prusa.connect.client import PrusaConnectClient client = PrusaConnectClient() -cameras = client.get_cameras() +cameras = client.cameras.list() if cameras: cam = cameras[0] @@ -54,10 +54,10 @@ List files on your team's storage. from prusa.connect.client import PrusaConnectClient client = PrusaConnectClient() -teams = client.get_teams() +teams = client.teams.list_teams() if teams: my_team_id = teams[0].id - files = client.get_file_list(my_team_id) + files = client.files.list(my_team_id) for file in files: print(f"{file.name} ({file.size.human_readable() if file.size else 'N/A'})") diff --git a/docs/sdk/quickstart.md b/docs/sdk/quickstart.md index c8d96ea..21caf2e 100644 --- a/docs/sdk/quickstart.md +++ b/docs/sdk/quickstart.md @@ -61,7 +61,7 @@ from prusa.connect.client import PrusaConnectClient client = PrusaConnectClient() print("My Printers:") -for printer in client.get_printers(): +for printer in client.printers.list_printers(): status = printer.printer_state or "UNKNOWN" print(f"- {printer.name} ({status})") @@ -86,7 +86,7 @@ from prusa.connect.client.exceptions import PrusaApiError, PrusaNetworkError client = PrusaConnectClient() try: - printers = client.get_printers() + printers = client.printers.list_printers() except PrusaApiError as e: # HTTP error from the Prusa Connect API (4xx / 5xx) print(f"API error {e.status_code}: {e}") diff --git a/mkdocs.yml b/mkdocs.yml index 59b412b..1dbc9bd 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -89,7 +89,7 @@ nav: - Client: api/client.md - Models: api/models.md - Development Resources: - - Contributing: contributing.md + - Contributing: CONTRIBUTING.md - Architecture Notes: internal_architecture.md not_in_nav: | /api/index.md diff --git a/pyproject.toml b/pyproject.toml index 5f5bbef..d5a5ad3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,12 +1,12 @@ [project] -name = "prusa-connect-sdk-client" # Normalized per https://packaging.python.org/en/latest/specifications/name-normalization/ +name = "prusa-connect-sdk-client" description = "Unoriginal Prusa Connect API client for Python and CLI" readme = "README.md" authors = [ { name = "Derek Ditch", email = "dcode@users.noreply.github.com"} ] classifiers = [ - "Development Status :: 4 - Beta", + "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)", "Programming Language :: Python :: 3", @@ -108,7 +108,7 @@ packages = ["src/prusa"] [tool.hatch.envs.default] installer = "uv" -env-vars = { "UV_DEFAULT_INDEX" = "https://pypi.org/simple/" } +env-vars = { "UV_DEFAULT_INDEX" = "https://pypi.org/simple" } features = ["cli"] [tool.hatch.build.hooks.protobuf] diff --git a/src/prusa/connect/client/__version__.py b/src/prusa/connect/client/__version__.py index a3d76cc..7772031 100644 --- a/src/prusa/connect/client/__version__.py +++ b/src/prusa/connect/client/__version__.py @@ -9,4 +9,4 @@ ``` """ -__version__ = "1.0.0a2" +__version__ = "1.0.0" diff --git a/src/prusa/connect/client/cli/commands/api.py b/src/prusa/connect/client/cli/commands/api.py index 613edce..f69d420 100644 --- a/src/prusa/connect/client/cli/commands/api.py +++ b/src/prusa/connect/client/cli/commands/api.py @@ -9,7 +9,6 @@ import cyclopts import requests # noqa: TC002 -from rich import print as rprint from prusa.connect.client.cli import common @@ -49,10 +48,10 @@ def api_command( res: requests.Response = client._request(method, path, raw=True, **kwargs) if response_headers: - rprint(f"{getattr(res, 'status_code', None)} {getattr(res, 'reason', None)}") + print(f"{getattr(res, 'status_code', None)} {getattr(res, 'reason', None)}") for k, v in res.headers.items(): - rprint(f"[bold]{k}:[/bold] {v}") - rprint("") + print(f"{k}: {v}") + print("") if stream: # Handle streaming @@ -60,7 +59,7 @@ def api_command( with open(output, "wb") as f: for chunk in res.iter_content(chunk_size=8192): f.write(chunk) - rprint(f"[green]Streamed response to {output}[/green]") + common.output_message(f"Streamed response to {output}") else: # Stream to stdout for chunk in res.iter_content(chunk_size=8192): @@ -88,7 +87,7 @@ def api_command( f.write(res.text) else: output.write_bytes(res.content) - rprint(f"[green]Response saved to {output}[/green]") + common.output_message(f"Response saved to {output}") else: if response_body: if "application/json" in content_type.lower(): @@ -103,4 +102,4 @@ def api_command( # If piping or streaming, print error to stderr sys.stderr.write(f"API Request Failed: {e}\n") else: - rprint(f"[red]API Request Failed: {e}[/red]") + common.output_message(f"API Request Failed: {e}", error=True) diff --git a/src/prusa/connect/client/cli/commands/auth.py b/src/prusa/connect/client/cli/commands/auth.py index 90542af..ae43e9f 100644 --- a/src/prusa/connect/client/cli/commands/auth.py +++ b/src/prusa/connect/client/cli/commands/auth.py @@ -8,9 +8,7 @@ import typing import cyclopts -from rich import print as rprint from rich.prompt import Confirm, Prompt -from rich.table import Table from prusa.connect.client import auth, exceptions from prusa.connect.client.cli import common, config @@ -21,14 +19,15 @@ @auth_app.command(name="login") def login_command(): """Perform interactive login.""" - rprint("[bold blue]Logging in to Prusa Connect...[/bold blue]") + # Always use rich console for interactive prompts regardless of format + common.console.print("[bold blue]Logging in to Prusa Connect...[/bold blue]") default_email = config.settings.prusa_email or os.environ.get("PRUSA_EMAIL") email = Prompt.ask("Email", default=default_email) default_password = config.settings.prusa_password or os.environ.get("PRUSA_PASSWORD") if default_password: - rprint("[dim]Using password from environment/config[/dim]") + common.console.print("[dim]Using password from environment/config[/dim]") password = default_password else: password = Prompt.ask("Password", password=True) @@ -41,19 +40,17 @@ def otp_callback() -> str: def save_tokens(data): path = config.settings.tokens_file - # Ensure parent dir exists path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w") as f: json.dump(data, f, indent=2) - # Best-effort secure permissions with contextlib.suppress(OSError): os.chmod(path, 0o600) save_tokens(token_data.dump_tokens()) - rprint(f"[green]Authentication successful! Tokens saved to {config.settings.tokens_file}[/green]") + common.output_message(f"Authentication successful! Tokens saved to {config.settings.tokens_file}") except Exception as e: - rprint(f"[bold red]Authentication failed: {e}[/bold red]") + common.output_message(f"Authentication failed: {e}", error=True) sys.exit(1) @@ -66,42 +63,81 @@ def show_command(): if creds: creds.refresh() except exceptions.PrusaAuthError: - rprint("[yellow]Not authenticated or tokens expired.[/yellow]") + common.output_message("Not authenticated or tokens expired.") return if not creds or not creds.tokens: - rprint("[yellow]No tokens found.[/yellow]") + common.output_message("No tokens found.") return t = creds.tokens - table = Table(title="Authentication Status", show_header=False) - # helper def fmt_ts(ts): return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") - if t.identity_token: - table.add_row("[bold]Identity[/bold]", "") - table.add_row(" Subject (Sub)", str(t.identity_token.user_id)) - table.add_row(" Issuer (Iss)", t.identity_token.issuer) - table.add_row(" Token ID (JTI)", t.identity_token.token_id) - if t.identity_token.user_info: - for k, v in t.identity_token.user_info.items(): - if v: - table.add_row(f" User.{k}", str(v)) - - if t.access_token: - table.add_row("[bold]Access Token[/bold]", "") - table.add_row(" Token ID (JTI)", t.access_token.token_id) - table.add_row(" Expires", fmt_ts(t.access_token.expires_at.timestamp())) - table.add_row(" Scope", ", ".join(t.scope)) - - if t.refresh_token: - table.add_row("[bold]Refresh Token[/bold]", "") - table.add_row(" Token ID (JTI)", t.refresh_token.token_id) - table.add_row(" Expires", fmt_ts(t.refresh_token.expires_at.timestamp())) - - common.console.print(table) + fmt = common.get_output_format() + + if fmt == "json": + # Serialize structured token data + data: dict[str, dict[str, str | list[str] | dict[str, str]]] = {} + if t.identity_token: + identity: dict[str, str | list[str] | dict[str, str]] = { + "user_id": str(t.identity_token.user_id), + "issuer": t.identity_token.issuer, + "token_id": t.identity_token.token_id, + } + if t.identity_token.user_info: + identity["user_info"] = {k: str(v) for k, v in t.identity_token.user_info.items() if v} + data["identity"] = identity + if t.access_token: + data["access_token"] = { + "token_id": t.access_token.token_id, + "expires": fmt_ts(t.access_token.expires_at.timestamp()), + "scope": list(t.scope), + } + if t.refresh_token: + data["refresh_token"] = { + "token_id": t.refresh_token.token_id, + "expires": fmt_ts(t.refresh_token.expires_at.timestamp()), + } + print(json.dumps(data)) + else: + rows: list[list[str]] = [] + + if t.identity_token: + rows.append(["Identity", ""]) + rows.append([" Subject (Sub)", str(t.identity_token.user_id)]) + rows.append([" Issuer (Iss)", t.identity_token.issuer]) + rows.append([" Token ID (JTI)", t.identity_token.token_id]) + if t.identity_token.user_info: + for k, v in t.identity_token.user_info.items(): + if v: + rows.append([f" User.{k}", str(v)]) + + if t.access_token: + rows.append(["Access Token", ""]) + rows.append([" Token ID (JTI)", t.access_token.token_id]) + rows.append([" Expires", fmt_ts(t.access_token.expires_at.timestamp())]) + rows.append([" Scope", ", ".join(t.scope)]) + + if t.refresh_token: + rows.append(["Refresh Token", ""]) + rows.append([" Token ID (JTI)", t.refresh_token.token_id]) + rows.append([" Expires", fmt_ts(t.refresh_token.expires_at.timestamp())]) + + if fmt == "plain": + print("# Authentication Status") + for label, value in rows: + print(f"{label}\t{value}") + else: + from rich.table import Table + + table = Table(title="Authentication Status", show_header=False) + table.add_column("Key", style="bold") + table.add_column("Value") + for label, value in rows: + table.add_row(label, value) + common.console.print(table) @auth_app.command(name="clear") @@ -110,24 +146,23 @@ def clear_command(): path = config.settings.tokens_file if path.exists(): if not Confirm.ask(f"Clear saved credentials at {path}?"): - rprint("[dim]Aborted.[/dim]") + common.output_message("Aborted.") return path.unlink() - rprint(f"[green]Removed tokens file: {path}[/green]") + common.output_message(f"Removed tokens file: {path}") else: - rprint(f"[yellow]No tokens file found at {path}[/yellow]") + common.output_message(f"No tokens file found at {path}") def _print_token(kind: typing.Literal["access", "identity"]): """Helper to print raw token.""" creds = auth.PrusaConnectCredentials.load_default() - # Try refresh if needed if creds and not creds.valid: with contextlib.suppress(exceptions.PrusaAuthError): creds.refresh() if not creds or not creds.tokens: - rprint("[red]No credentials found.[/red]", file=sys.stderr) + common.output_message("No credentials found.", error=True) sys.exit(1) token = None @@ -139,7 +174,7 @@ def _print_token(kind: typing.Literal["access", "identity"]): if token: print(token) else: - rprint(f"[red]No {kind} token found.[/red]", file=sys.stderr) + common.output_message(f"No {kind} token found.", error=True) sys.exit(1) @@ -153,7 +188,3 @@ def print_access_token_command(): def print_identity_token_command(): """Print the raw identity token.""" _print_token("identity") - - -# Legacy alias for backward compatibility if needed, but we're replacing the command structure. -# We can export auth_app as the main interface. diff --git a/src/prusa/connect/client/cli/commands/camera.py b/src/prusa/connect/client/cli/commands/camera.py index 614bbb0..e1b754c 100644 --- a/src/prusa/connect/client/cli/commands/camera.py +++ b/src/prusa/connect/client/cli/commands/camera.py @@ -5,13 +5,17 @@ import typing import cyclopts -from rich import print as rprint -from rich.table import Table from prusa.connect.client.cli import common, config camera_app = cyclopts.App(name="camera", help="Camera management") +_NO_CAMERA = ( + "No camera ID provided and no default configured.\n" + "Hint: Run 'prusactl camera list' to find a numeric ID, then " + "'prusactl camera set-current ' to set the default." +) + @camera_app.command(name="list") def camera_list(): @@ -20,41 +24,39 @@ def camera_list(): client = common.get_client() cameras = client.cameras.list() - table = Table(title="Cameras") - table.add_column("Name", style="cyan") - table.add_column("ID (Numeric)", style="magenta") - table.add_column("Token", style="green") - table.add_column("Origin", style="blue") - - for c in cameras: - table.add_row( - c.name or "Unknown", - str(c.id) if c.id else "N/A", - c.token or "N/A", - c.origin or "N/A", - ) - common.console.print(table) + rows = [[c.name or "Unknown", str(c.id) if c.id else "N/A", c.token or "N/A", c.origin or "N/A"] for c in cameras] + common.output_table( + "Cameras", + ["Name", "ID (Numeric)", "Token", "Origin"], + rows, + column_styles=["cyan", "magenta", "green", "blue"], + ) -def camera_alias(): +def cameras_alias(): """List all cameras (alias for 'camera list').""" camera_list() @camera_app.command(name="snapshot") def camera_snapshot( - camera_id: typing.Annotated[str, cyclopts.Parameter(help="Camera ID (Numeric)")], + camera_id: typing.Annotated[str | None, cyclopts.Parameter(help="Camera ID (Numeric)")] = None, output: typing.Annotated[pathlib.Path | None, cyclopts.Parameter(help="Output file path for snapshot")] = None, ): """Take a snapshot from a camera.""" common.logger.debug("Command started", command="camera snapshot", camera_id=camera_id, output=output) client = common.get_client() + resolved_id = camera_id or config.settings.default_camera_id + if not resolved_id: + common.output_message(_NO_CAMERA, error=True) + return + # We look up the camera to get ID cameras = client.cameras.list() match = next((c for c in cameras if str(c.id) == camera_id or c.token == camera_id or c.name == camera_id), None) - real_id = camera_id + real_id = resolved_id if match: if match.id: real_id = str(match.id) @@ -67,29 +69,36 @@ def camera_snapshot( sys.stdout.buffer.write(data) else: output.write_bytes(data) - rprint(f"[green]Saved snapshot to {output}[/green]") + common.output_message(f"Saved snapshot to {output}") else: - rprint(f"[green]Snapshot received: {len(data)} bytes[/green]") + common.output_message(f"Snapshot received: {len(data)} bytes") except Exception as e: if str(output) == "-": sys.stderr.write(f"Failed to get snapshot: {e}\n") else: - rprint(f"[red]Failed to get snapshot: {e}[/red]") + common.output_message(f"Failed to get snapshot: {e}", error=True) @camera_app.command(name="trigger") def camera_trigger( - camera_id: typing.Annotated[str, cyclopts.Parameter(help="Camera Token or ID (if mapped)")], + camera_id: typing.Annotated[str | None, cyclopts.Parameter(help="Camera Token or ID (if mapped)")] = None, ): """Trigger a snapshot on a camera.""" common.logger.debug("Command started", command="camera trigger", camera_id=camera_id) client = common.get_client() + resolved_id = camera_id or config.settings.default_camera_id + if not resolved_id: + common.output_message(_NO_CAMERA, error=True) + return + # We look up to get token cameras = client.cameras.list() - match = next((c for c in cameras if str(c.id) == camera_id or c.token == camera_id or c.name == camera_id), None) + match = next( + (c for c in cameras if str(c.id) == resolved_id or c.token == resolved_id or c.name == resolved_id), None + ) - real_token = camera_id + real_token = resolved_id if match: if match.token: real_token = match.token @@ -97,39 +106,52 @@ def camera_trigger( try: if client.trigger_snapshot(real_token): - rprint(f"[green]Triggered snapshot for {camera_id}[/green]") + common.output_message(f"Triggered snapshot for {camera_id}") except Exception as e: - rprint(f"[red]Failed to trigger snapshot: {e}[/red]") + common.output_message(f"Failed to trigger snapshot: {e}", error=True) @camera_app.command(name="move") def camera_move( - camera_id: typing.Annotated[str, cyclopts.Parameter(help="Camera Token or ID")], - direction: typing.Annotated[str, cyclopts.Parameter(help="Direction: LEFT, RIGHT, UP, DOWN")], + camera_id: typing.Annotated[str | None, cyclopts.Parameter(help="Camera Token or ID")] = None, + direction: typing.Annotated[ + typing.Literal["LEFT", "RIGHT", "UP", "DOWN"] | None, + cyclopts.Parameter(help="Direction: LEFT, RIGHT, UP, DOWN"), + ] = None, angle: typing.Annotated[int, cyclopts.Parameter(help="Angle in degrees")] = 30, ): """Move a pan/tilt camera.""" common.logger.debug("Command started", command="camera move", camera_id=camera_id, direction=direction) client = common.get_client() + resolved_id = camera_id or config.settings.default_camera_id + if not resolved_id: + common.output_message(_NO_CAMERA, error=True) + return + + if direction is None: + common.output_message("A direction (LEFT, RIGHT, UP, or DOWN) must be specified", error=True) + # Resolve token cameras = client.cameras.list() - match = next((c for c in cameras if str(c.id) == camera_id or c.token == camera_id or c.name == camera_id), None) - token = match.token if match and match.token else camera_id + match = next( + (c for c in cameras if str(c.id) == resolved_id or c.token == resolved_id or c.name == resolved_id), None + ) + token = match.token if match and match.token else resolved_id try: cam_client = client.get_camera_client(token) cam_client.connect() - cam_client.move(direction, angle) - rprint(f"[green]Sent {direction} move command to {camera_id}[/green]") + cam_client.move(str(direction), angle) + common.output_message(f"Sent {direction} move command to {camera_id}") cam_client.disconnect() except Exception as e: - rprint(f"[red]Failed to move camera: {e}[/red]") + common.output_message(f"Failed to move camera: {e}", error=True) @camera_app.command(name="adjust") def camera_adjust( - camera_id: typing.Annotated[str, cyclopts.Parameter(help="Camera Token or ID")], + camera_id: typing.Annotated[str | None, cyclopts.Parameter(help="Camera Token or ID")] = None, brightness: typing.Annotated[int | None, cyclopts.Parameter(help="Brightness value")] = None, contrast: typing.Annotated[int | None, cyclopts.Parameter(help="Contrast value")] = None, saturation: typing.Annotated[int | None, cyclopts.Parameter(help="Saturation value")] = None, @@ -138,10 +160,17 @@ def camera_adjust( common.logger.debug("Command started", command="camera adjust", camera_id=camera_id) client = common.get_client() + resolved_id = camera_id or config.settings.default_camera_id + if not resolved_id: + common.output_message(_NO_CAMERA, error=True) + return + # Resolve token cameras = client.cameras.list() - match = next((c for c in cameras if str(c.id) == camera_id or c.token == camera_id or c.name == camera_id), None) - token = match.token if match and match.token else camera_id + match = next( + (c for c in cameras if str(c.id) == resolved_id or c.token == resolved_id or c.name == resolved_id), None + ) + token = match.token if match and match.token else resolved_id kwargs = {} if brightness is not None: @@ -152,17 +181,17 @@ def camera_adjust( kwargs["saturation"] = saturation if not kwargs: - rprint("[yellow]No adjustments provided. Use --brightness, --contrast, or --saturation.[/yellow]") + common.output_message("No adjustments provided. Use --brightness, --contrast, or --saturation.") return try: cam_client = client.get_camera_client(token) cam_client.connect() cam_client.adjust(**kwargs) - rprint(f"[green]Adjusted settings for {camera_id}[/green]") + common.output_message(f"Adjusted settings for {camera_id}") cam_client.disconnect() except Exception as e: - rprint(f"[red]Failed to adjust camera: {e}[/red]") + common.output_message(f"Failed to adjust camera: {e}", error=True) @camera_app.command(name="set-current") @@ -170,49 +199,59 @@ def set_current_camera(camera_id: typing.Annotated[str, cyclopts.Parameter(help= """Set the default camera ID for future commands.""" config.settings.default_camera_id = camera_id config.save_json_config(config.settings) - rprint(f"[green]Successfully set default camera to {camera_id}[/green]") + common.output_message(f"Successfully set default camera to {camera_id}") @camera_app.command(name="show") def camera_show( - camera_id: typing.Annotated[str, cyclopts.Parameter(help="Camera Token or ID or Name")], + camera_id: typing.Annotated[str | None, cyclopts.Parameter(help="Camera Token or ID or Name")] = None, detailed: bool = False, ): """Show details for a specific camera.""" common.logger.debug("Command started", command="camera show", camera_id=camera_id, detailed=detailed) client = common.get_client() + resolved_id = camera_id or config.settings.default_camera_id + if not resolved_id: + common.output_message(_NO_CAMERA, error=True) + return + cameras = client.cameras.list() - match = next((c for c in cameras if str(c.id) == camera_id or c.token == camera_id or c.name == camera_id), None) + match = next( + (c for c in cameras if str(c.id) == resolved_id or c.token == resolved_id or c.name == resolved_id), None + ) if not match: - rprint(f"[red]Camera '{camera_id}' not found.[/red]") + common.output_message(f"Camera '{camera_id}' not found.", error=True) sys.exit(1) - if detailed: + if detailed and common.get_output_format() == "rich": from rich.panel import Panel from rich.pretty import Pretty common.console.print(Panel(Pretty(match), title=f"Camera: {match.name or 'Unknown'}")) else: - table = Table(show_header=False, box=None) - table.add_column("Property", style="bold cyan") - table.add_column("Value") - - table.add_row("Name", match.name or "N/A") - table.add_row("ID (Numeric)", str(match.id) if match.id else "N/A") - table.add_row("Token", match.token or "N/A") - table.add_row("Origin", match.origin or "N/A") + rows = [ + ["Name", match.name or "N/A"], + ["ID (Numeric)", str(match.id) if match.id else "N/A"], + ["Token", match.token or "N/A"], + ["Origin", match.origin or "N/A"], + ] if match.config: if match.config.resolution: - table.add_row("Resolution", f"{match.config.resolution.width}x{match.config.resolution.height}") + rows.append(["Resolution", f"{match.config.resolution.width}x{match.config.resolution.height}"]) if match.config.firmware: - table.add_row("Firmware", match.config.firmware) + rows.append(["Firmware", match.config.firmware]) if match.config.model: - table.add_row("Model", match.config.model) + rows.append(["Model", match.config.model]) if match.printer_uuid: - table.add_row("Printer UUID", match.printer_uuid) + rows.append(["Printer UUID", match.printer_uuid]) - common.console.print(table) + common.output_table( + f"Camera: {match.name or 'Unknown'}", + ["Property", "Value"], + rows, + column_styles=["bold cyan", None], + ) diff --git a/src/prusa/connect/client/cli/commands/file.py b/src/prusa/connect/client/cli/commands/file.py index 8e47132..127d166 100644 --- a/src/prusa/connect/client/cli/commands/file.py +++ b/src/prusa/connect/client/cli/commands/file.py @@ -1,10 +1,10 @@ """File management commands.""" +import json import os import typing import cyclopts -from rich.table import Table from prusa.connect.client.cli import common, config @@ -22,26 +22,24 @@ def file_list( if not resolved_team_id: teams = client.teams.list_teams() if not teams: - common.console.print("[red]No teams found.[/red]") + common.output_message("No teams found.", error=True) return resolved_team_id = teams[0].id - common.console.print( - f"[yellow]No team ID provided. Using first team: {teams[0].name} ({resolved_team_id})[/yellow]" - ) + common.output_message(f"No team ID provided. Using first team: {teams[0].name} ({resolved_team_id})") files = client.get_file_list(resolved_team_id) - table = Table(title=f"Files for Team {resolved_team_id}") - table.add_column("Name", style="cyan") - table.add_column("Type", style="green") - table.add_column("Size", style="magenta") - table.add_column("Hash", style="blue") - + rows = [] for f in files: size_str = f"{f.size.human_readable()}" if f.size else "N/A" - table.add_row(f.name or "N/A", f.type or "N/A", size_str, f.hash or "N/A") + rows.append([f.name or "N/A", f.type or "N/A", size_str, f.hash or "N/A"]) - common.console.print(table) + common.output_table( + f"Files for Team {resolved_team_id}", + ["Name", "Type", "Size", "Hash"], + rows, + column_styles=["cyan", "green", "magenta", "blue"], + ) def files_alias( @@ -63,23 +61,23 @@ def file_upload( if not resolved_team_id: teams = client.teams.list_teams() if not teams: - common.console.print("[red]No teams found.[/red]") + common.output_message("No teams found.", error=True) return resolved_team_id = teams[0].id file_path = os.path.abspath(path) if not os.path.exists(file_path): - common.console.print(f"[red]File not found: {path}[/red]") + common.output_message(f"File not found: {path}", error=True) return filename = os.path.basename(file_path) size = os.path.getsize(file_path) - common.console.print(f"Initiating upload for [cyan]{filename}[/cyan] ({size} bytes)...") + common.output_message(f"Initiating upload for {filename} ({size} bytes)...") try: status = client.initiate_team_upload(resolved_team_id, destination, filename, size) upload_id = status.id - common.console.print(f"Upload initiated. ID: [magenta]{upload_id}[/magenta]. Uploading data...") + common.output_message(f"Upload initiated. ID: {upload_id}. Uploading data...") with open(file_path, "rb") as f: data = f.read() @@ -91,9 +89,9 @@ def file_upload( content_type = "text/x.gcode" client.upload_team_file(resolved_team_id, upload_id, data, content_type=content_type) - common.console.print("[green]Upload successful![/green]") + common.output_message("Upload successful!") except Exception as e: - common.console.print(f"[red]Upload failed: {e}[/red]") + common.output_message(f"Upload failed: {e}", error=True) @file_app.command(name="download") @@ -108,11 +106,11 @@ def file_download( if not resolved_team_id: teams = client.teams.list_teams() if not teams: - common.console.print("[red]No teams found.[/red]") + common.output_message("No teams found.", error=True) return resolved_team_id = teams[0].id - common.console.print(f"Downloading file with hash [cyan]{file_hash}[/cyan]...") + common.output_message(f"Downloading file with hash {file_hash}...") try: data = client.download_team_file(resolved_team_id, file_hash) @@ -120,9 +118,9 @@ def file_download( with open(dest_path, "wb") as f: f.write(data) - common.console.print(f"[green]Downloaded to {dest_path}[/green]") + common.output_message(f"Downloaded to {dest_path}") except Exception as e: - common.console.print(f"[red]Download failed: {e}[/red]") + common.output_message(f"Download failed: {e}", error=True) @file_app.command(name="show") @@ -139,38 +137,40 @@ def file_show( if not resolved_team_id: teams = client.teams.list_teams() if not teams: - common.console.print("[red]No teams found.[/red]") + common.output_message("No teams found.", error=True) return resolved_team_id = teams[0].id try: file = client.get_team_file(resolved_team_id, file_hash) - table = Table(title=f"File Details: {file.name}") - table.add_column("Property", style="cyan") - table.add_column("Value", style="green") - - table.add_row("Name", file.name) - table.add_row("Type", getattr(file, "type", "N/A")) + rows: list[list[str]] = [["Name", file.name], ["Type", getattr(file, "type", "N/A")]] if getattr(file, "size", None) is not None: size_val = typing.cast("int", file.size) - size_str = f"{size_val / 1024 / 1024:.2f} MB" - table.add_row("Size", size_str) + rows.append(["Size", f"{size_val / 1024 / 1024:.2f} MB"]) if getattr(file, "hash", None): - table.add_row("Hash", file.hash) + rows.append(["Hash", file.hash]) - common.console.print(table) + common.output_table( + f"File Details: {file.name}", + ["Property", "Value"], + rows, + column_styles=["cyan", "green"], + ) if detailed: - import json - - common.console.print("\n[bold]Detailed Information:[/bold]") - detail_table = Table(show_header=False, box=None) + detail_rows = [] for k, v in file.model_dump(mode="json").items(): if v is not None and k not in ["name", "type", "size", "hash"]: val_str = json.dumps(v) if isinstance(v, (dict, list)) else str(v) - detail_table.add_row(f"[cyan]{k.title().replace('_', ' ')}[/cyan]:", val_str) - common.console.print(detail_table) + detail_rows.append([k.title().replace("_", " "), val_str]) + if detail_rows: + common.output_table( + "Detailed Information", + ["Field", "Value"], + detail_rows, + column_styles=["cyan", None], + ) except Exception as e: - common.console.print(f"[red]Failed to fetch file details: {e}[/red]") + common.output_message(f"Failed to fetch file details: {e}", error=True) diff --git a/src/prusa/connect/client/cli/commands/job.py b/src/prusa/connect/client/cli/commands/job.py index b70910f..808bed8 100644 --- a/src/prusa/connect/client/cli/commands/job.py +++ b/src/prusa/connect/client/cli/commands/job.py @@ -4,8 +4,6 @@ import typing import cyclopts -from rich import print as rprint -from rich.table import Table from prusa.connect.client.cli import common, config @@ -38,15 +36,12 @@ def job_list( all_jobs.extend(client.get_team_jobs(team, state=state, limit=limit)) else: # Aggregation mode: Get jobs from ALL printers (cached) - # This is preferred over iterating teams if we want "my printers" context try: printers = client.printers.list_printers() for p in printers: if not p.uuid: continue try: - # We fetch 'limit' items from EACH printer to ensure we have enough candidates for global sort - # If limit is None, we fetch default page p_jobs = client.get_printer_jobs(p.uuid, state=state, limit=limit) all_jobs.extend(p_jobs) except Exception as e: @@ -55,44 +50,39 @@ def job_list( common.logger.error("Failed to fetch printer list for aggregation", error=str(e)) # Sort by 'end' time (descending) to show most recent first - # Fallback to 'start' or 'id' if 'end' is missing def sort_key(j): - # We want descending order, so we return a tuple that compares correctly - # Use 0 as fallback for timestamps if missing return (j.end or 0, j.start or 0, j.id or 0) all_jobs.sort(key=sort_key, reverse=True) - # Apply global limit if limit is not None: all_jobs = all_jobs[:limit] - table = Table(title="Jobs") - table.add_column("ID", style="cyan") - table.add_column("Printer", style="magenta") - table.add_column("State", style="green") - table.add_column("Name", style="blue") - table.add_column("Progress", style="yellow") - table.add_column("Ended", style="dim") - + rows = [] for j in all_jobs: - # Format timestamp ended_str = "N/A" if j.end: ended_str = datetime.datetime.fromtimestamp(j.end).strftime("%Y-%m-%d %H:%M") elif j.state == "PRINTING": ended_str = "In Progress" - table.add_row( - str(j.id), - j.printer_uuid or "Unknown", - j.state.name, - j.file.name if j.file else "Unknown", - f"{j.progress}%" if j.progress is not None else "N/A", - ended_str, + rows.append( + [ + str(j.id), + j.printer_uuid or "Unknown", + j.state.name, + j.file.name if j.file else "Unknown", + f"{j.progress}%" if j.progress is not None else "N/A", + ended_str, + ] ) - common.console.print(table) + common.output_table( + "Jobs", + ["ID", "Printer", "State", "Name", "Progress", "Ended"], + rows, + column_styles=["cyan", "magenta", "green", "blue", "yellow", "dim"], + ) def jobs_alias( @@ -119,7 +109,7 @@ def job_queued( try: all_jobs.extend(client.get_printer_queue(printer)) except Exception as e: - rprint(f"[red]Failed to fetch queue for {printer}: {e}[/red]") + common.output_message(f"Failed to fetch queue for {printer}: {e}", error=True) else: # Aggregate from all printers try: @@ -133,31 +123,26 @@ def job_queued( except Exception as e: common.logger.warning(f"Failed to fetch queue for printer {p.name}", error=str(e)) except Exception as e: - rprint(f"[red]Failed to fetch printer list: {e}[/red]") - - # Sort by creation/id (ascending for queue usually? or purely by order returned?) - # Usually queues are FIFO, but aggregation might mix them. - # We'll trust the order or sort by ID/date if available. - # For now, let's keep them somewhat creation-ordered if possible. + common.output_message(f"Failed to fetch printer list: {e}", error=True) - table = Table(title="Job Queue") - table.add_column("ID", style="cyan") - table.add_column("Printer", style="magenta") - table.add_column("State", style="green") - table.add_column("Name", style="blue") - table.add_column("Source", style="dim") + if not all_jobs: + common.output_message("No jobs in queue.") + return + rows = [] for j in all_jobs: source = "Unknown" if j.source_info: source = j.source_info.public_name or j.source_info.first_name or "Unknown" - table.add_row(str(j.id), j.printer_uuid or "Unknown", j.state, j.file.name if j.file else "Unknown", source) + rows.append([str(j.id), j.printer_uuid or "Unknown", j.state, j.file.name if j.file else "Unknown", source]) - if not all_jobs: - rprint("[yellow]No jobs in queue.[/yellow]") - else: - common.console.print(table) + common.output_table( + "Job Queue", + ["ID", "Printer", "State", "Name", "Source"], + rows, + column_styles=["cyan", "magenta", "green", "blue", "dim"], + ) @job_app.command(name="show") @@ -173,40 +158,42 @@ def job_show( client = common.get_client() resolved_printer_id = printer or config.settings.default_printer_id if not resolved_printer_id: - rprint("[red]Printer UUID is required. Provide --printer or set a default.[/red]") + common.output_message("Printer UUID is required. Provide --printer or set a default.", error=True) return try: job = client.get_job(resolved_printer_id, job_id) - # Basic Info - rprint(f"[bold cyan]Job {job.id}[/bold cyan]") - rprint(f"State: [green]{job.state}[/green]") + rows = [["ID", str(job.id)], ["State", str(job.state)]] if job.file: - rprint(f"File: {job.file.name}") + rows.append(["File", job.file.name]) if job.progress is not None: - rprint(f"Progress: {job.progress}%") + rows.append(["Progress", f"{job.progress}%"]) if job.time_printing: - rprint(f"Time Printing: {job.time_printing}s") + rows.append(["Time Printing", str(job.time_printing)]) + + common.output_table( + f"Job {job.id}", + ["Field", "Value"], + rows, + column_styles=["cyan", None], + ) # Cancelable Objects if job.cancelable_objects: - rprint("\n[bold]Cancelable Objects:[/bold]") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("ID", style="cyan", justify="right") - table.add_column("Name", style="white") - - for obj in job.cancelable_objects: - table.add_row(str(obj.id), obj.name) - - common.console.print(table) + obj_rows = [[str(obj.id), obj.name] for obj in job.cancelable_objects] + common.output_table( + "Cancelable Objects", + ["ID", "Name"], + obj_rows, + column_styles=["cyan", "white"], + ) else: - rprint("\n[dim]No cancelable objects found for this job.[/dim]") + common.output_message("No cancelable objects found for this job.") if detailed: import json - rprint("\n[bold]Detailed Information:[/bold]") - detail_table = Table(show_header=False, box=None) + detail_rows = [] for k, v in job.model_dump(mode="json").items(): if v is not None and k not in [ "id", @@ -217,8 +204,15 @@ def job_show( "cancelable_objects", ]: val_str = json.dumps(v) if isinstance(v, (dict, list)) else str(v) - detail_table.add_row(f"[cyan]{k.title().replace('_', ' ')}[/cyan]:", val_str) - common.console.print(detail_table) + detail_rows.append([k.title().replace("_", " "), val_str]) + + if detail_rows: + common.output_table( + "Detailed Information", + ["Field", "Value"], + detail_rows, + column_styles=["cyan", None], + ) except Exception as e: - rprint(f"[red]Failed to fetch job details: {e}[/red]") + common.output_message(f"Failed to fetch job details: {e}", error=True) diff --git a/src/prusa/connect/client/cli/commands/printer.py b/src/prusa/connect/client/cli/commands/printer.py index 8c88e20..3e85ed3 100644 --- a/src/prusa/connect/client/cli/commands/printer.py +++ b/src/prusa/connect/client/cli/commands/printer.py @@ -6,8 +6,6 @@ import typing import cyclopts -from rich import print as rprint -from rich.table import Table from prusa.connect.client import exceptions, models from prusa.connect.client.cli import common, config @@ -16,6 +14,12 @@ files_printer_app = cyclopts.App(name="files", help="Printer file management") printer_app.command(files_printer_app) +_NO_PRINTER = ( + "No printer ID provided and no default configured.\n" + "Hint: Run 'prusactl printer list' to find a UUID, then " + "'prusactl printer set-current ' to set the default." +) + def _send_printer_command(printer_ids: list[str], command: str): """Helper to send a command to multiple printers.""" @@ -24,9 +28,9 @@ def _send_printer_command(printer_ids: list[str], command: str): for pid in printer_ids: try: if client.printers.send_command(pid, command): - rprint(f"[green]Sent {command} to {pid}[/green]") + common.output_message(f"Sent {command} to {pid}") except Exception as e: - rprint(f"[red]Failed to send {command} to {pid}: {e}[/red]") + common.output_message(f"Failed to send {command} to {pid}: {e}", error=True) @printer_app.command(name="list") @@ -39,25 +43,20 @@ def printer_list( printers = client.printers.list_printers() common.logger.info("Found printers", count=len(printers)) - table = Table(title="Printers") - table.add_column("Name", style="cyan") - table.add_column("UUID", style="magenta") - table.add_column("State", style="green") - table.add_column("Model", style="blue") - - # Filter filtered = [p for p in printers if fnmatch.fnmatch(p.name or "", pattern)] + rows = [] for p in filtered: common.logger.debug("Printer", json=p.model_dump_json()) state_str = str(p.printer_state) if p.printer_state else "UNKNOWN" - table.add_row( - p.name or "Unknown", - p.uuid or "Unknown", - state_str, - p.printer_model or "N/A", - ) - common.console.print(table) + rows.append([p.name or "Unknown", p.uuid or "Unknown", state_str, p.printer_model or "N/A"]) + + common.output_table( + "Printers", + ["Name", "UUID", "State", "Model"], + rows, + column_styles=["cyan", "magenta", "green", "blue"], + ) def printers_alias( @@ -77,11 +76,7 @@ def printer_show( """Show detailed status for a specific printer.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then\n" - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return common.logger.debug("Command started", command="printer show", printer_id=resolved_id) @@ -90,49 +85,42 @@ def printer_show( try: p = client.printers.get(resolved_id) - # Basic Info Table - table = Table(title=f"Printer: {p.name}") - table.add_column("Field", style="cyan") - table.add_column("Value", style="magenta") + # Build rows and track section boundaries + rows: list[list[str]] = [] + sections: set[int] = set() - table.add_row("UUID", p.uuid or "N/A") - table.add_row("State", p.printer_state or "N/A") - table.add_row("Model", p.printer_model or "N/A") + rows.append(["UUID", p.uuid or "N/A"]) + rows.append(["State", p.printer_state or "N/A"]) + rows.append(["Model", p.printer_model or "N/A"]) # Firmware fw_str = p.firmware_version or "Unknown" if p.support and p.support.latest and p.support.latest != p.firmware_version: - # Check if current != latest - # The 'current' field in support might be more accurate or redundant with p.firmware_version fw_str += f" [yellow](Latest: {p.support.latest})[/yellow]" - table.add_row("Firmware", fw_str) + rows.append(["Firmware", fw_str]) - # Location / Team if p.location: - table.add_row("Location", p.location) + rows.append(["Location", p.location]) if p.team_name: - table.add_row("Team", p.team_name) + rows.append(["Team", p.team_name]) # Network Info if p.network_info: - table.add_section() + sections.add(len(rows)) if p.network_info.hostname: - table.add_row("Hostname", p.network_info.hostname) + rows.append(["Hostname", p.network_info.hostname]) if p.network_info.lan_ipv4: - table.add_row("IP Address", p.network_info.lan_ipv4) + rows.append(["IP Address", p.network_info.lan_ipv4]) # Last Online if p.last_online: last_seen = datetime.datetime.fromtimestamp(p.last_online).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") - table.add_row("Last Online", last_seen) + rows.append(["Last Online", last_seen]) - # Tool 1 Material (Default View) + # Material material = "N/A" - # Try to find material from tools or slots if p.tools and "1" in p.tools: material = p.tools["1"].material or "N/A" - - # If no tool material, maybe checking active slot? if (material == "N/A" or material == "---") and p.slot and p.slot.active is not None: active_slot_key = str(p.slot.active) if p.slot.slots and active_slot_key in p.slot.slots: @@ -140,75 +128,82 @@ def printer_show( if m and m != "---": material = f"{m} (Slot {active_slot_key})" - table.add_section() - table.add_row("Material", material) + sections.add(len(rows)) + rows.append(["Material", material]) if p.telemetry: - table.add_row("Nozzle", f"{p.telemetry.temp_nozzle}°C") - table.add_row("Bed", f"{p.telemetry.temp_bed}°C") + rows.append(["Nozzle", f"{p.telemetry.temp_nozzle}°C"]) + rows.append(["Bed", f"{p.telemetry.temp_bed}°C"]) # Job if p.job: - table.add_section() - table.add_row("Job", p.job.display_name or "Unknown") - table.add_row("Progress", f"{p.job.progress}%") + sections.add(len(rows)) + rows.append(["Job", p.job.display_name or "Unknown"]) + rows.append(["Progress", f"{p.job.progress}%"]) if p.job.time_printing: - table.add_row("Time Printing", str(p.job.time_printing)) + rows.append(["Time Printing", str(p.job.time_printing)]) if p.job.time_remaining and p.job.time_remaining.total_seconds() > 0: - table.add_row("Time Remaining", str(p.job.time_remaining)) - - common.console.print(table) + rows.append(["Time Remaining", str(p.job.time_remaining)]) + + common.output_table( + f"Printer: {p.name}", + ["Field", "Value"], + rows, + column_styles=["cyan", "magenta"], + sections_before=sections, + ) if detailed: - # Detailed View - MMU Slots + # MMU Slots if p.slot and p.slot.slots: - slot_table = Table(title="MMU Slots") - slot_table.add_column("Slot", style="cyan") - slot_table.add_column("Material", style="magenta") - slot_table.add_column("Temp", style="yellow") - - # Sort by slot ID sorted_slots = sorted(p.slot.slots.items(), key=lambda x: int(x[0]) if x[0].isdigit() else 999) - for slot_id, slot_data in sorted_slots: - slot_table.add_row( - slot_id, slot_data.material or "---", str(slot_data.temp) if slot_data.temp else "N/A" - ) - common.console.print(slot_table) + slot_rows = [ + [slot_id, slot_data.material or "---", str(slot_data.temp) if slot_data.temp else "N/A"] + for slot_id, slot_data in sorted_slots + ] + common.output_table( + "MMU Slots", + ["Slot", "Material", "Temp"], + slot_rows, + column_styles=["cyan", "magenta", "yellow"], + ) - # Tools Detail (Fans etc) + # Tools if p.tools: - tool_table = Table(title="Tools / Heads") - tool_table.add_column("Tool", style="cyan") - tool_table.add_column("Nozzle", style="green") - tool_table.add_column("Material", style="magenta") - tool_table.add_column("Fan Print", style="blue") - tool_table.add_column("Fan Hotend", style="blue") - - for tool_id, tool_data in p.tools.items(): - tool_table.add_row( + tool_rows = [ + [ tool_id, str(tool_data.nozzle_diameter) if tool_data.nozzle_diameter else "N/A", tool_data.material or "---", f"{tool_data.fan_print}%" if tool_data.fan_print is not None else "N/A", f"{tool_data.fan_hotend}%" if tool_data.fan_hotend is not None else "N/A", - ) - common.console.print(tool_table) + ] + for tool_id, tool_data in p.tools.items() + ] + common.output_table( + "Tools / Heads", + ["Tool", "Nozzle", "Material", "Fan Print", "Fan Hotend"], + tool_rows, + column_styles=["cyan", "green", "magenta", "blue", "blue"], + ) - # Axis Info - axis_table = Table(title="Axis Positions") - axis_table.add_column("Axis", style="cyan") - axis_table.add_column("Position", style="yellow") + # Axis + axis_rows = [] if p.axis_x is not None: - axis_table.add_row("X", str(p.axis_x)) + axis_rows.append(["X", str(p.axis_x)]) if p.axis_y is not None: - axis_table.add_row("Y", str(p.axis_y)) + axis_rows.append(["Y", str(p.axis_y)]) if p.axis_z is not None: - axis_table.add_row("Z", str(p.axis_z)) - - if axis_table.row_count > 0: - common.console.print(axis_table) + axis_rows.append(["Z", str(p.axis_z)]) + if axis_rows: + common.output_table( + "Axis Positions", + ["Axis", "Position"], + axis_rows, + column_styles=["cyan", "yellow"], + ) - rprint("\n[bold]Raw Detailed Information:[/bold]") - detail_table = Table(show_header=False, box=None) + # Raw extra fields + detail_rows = [] for k, v in p.model_dump(mode="json").items(): if v is not None and k not in [ "uuid", @@ -227,11 +222,17 @@ def printer_show( "job", ]: val_str = json.dumps(v) if isinstance(v, (dict, list)) else str(v) - detail_table.add_row(f"[cyan]{k}[/cyan]:", val_str) - common.console.print(detail_table) + detail_rows.append([k, val_str]) + if detail_rows: + common.output_table( + "Raw Detailed Information", + ["Field", "Value"], + detail_rows, + column_styles=["cyan", None], + ) except exceptions.PrusaConnectError as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @printer_app.command(name="pause") @@ -241,7 +242,7 @@ def printer_pause( """Pause print on one or more printers.""" ids = printer_ids or ([config.settings.default_printer_id] if config.settings.default_printer_id else []) if not ids: - rprint("[red]No printer IDs provided and no default configured.[/red]") + common.output_message("No printer IDs provided and no default configured.", error=True) return common.logger.debug("Command started", command="printer pause", printer_ids=ids) @@ -255,7 +256,7 @@ def printer_resume( """Resume print on one or more printers.""" ids = printer_ids or ([config.settings.default_printer_id] if config.settings.default_printer_id else []) if not ids: - rprint("[red]No printer IDs provided and no default configured.[/red]") + common.output_message("No printer IDs provided and no default configured.", error=True) return common.logger.debug("Command started", command="printer resume", printer_ids=ids) @@ -271,7 +272,7 @@ def printer_stop( """Stop print on one or more printers, optionally setting a failure reason.""" ids = printer_ids or ([config.settings.default_printer_id] if config.settings.default_printer_id else []) if not ids: - rprint("[red]No printer IDs provided and no default configured.[/red]") + common.output_message("No printer IDs provided and no default configured.", error=True) return common.logger.debug("Command started", command="printer stop", printer_ids=ids, reason=reason) @@ -279,41 +280,32 @@ def printer_stop( for pid in ids: try: - # 1. Stop the print if client.stop_print(pid): - rprint(f"[green]Sent STOP_PRINT to {pid}[/green]") + common.output_message(f"Sent STOP_PRINT to {pid}") - # 2. Set reason if provided if reason: - # We need the current job ID to set the reason - # Fetch printer status to get job ID try: p = client.printers.get(pid) if p.job and p.job.id: - # Validate reason string against Enum - - # Accept a case insensitive reason string try: enum_reason = models.JobFailureTag(reason.upper()) client.set_job_failure_reason(pid, p.job.id, enum_reason, note) - rprint(f"[green]Set failure reason '{enum_reason}' for Job {p.job.id}[/green]") + common.output_message(f"Set failure reason '{enum_reason}' for Job {p.job.id}") except ValueError: - rprint( - f"[yellow]Invalid reason code '{reason}'. Supported: " - f"{', '.join([r.value for r in models.JobFailureTag])}[/yellow]" + common.output_message( + f"Invalid reason code '{reason}'. Supported: " + f"{', '.join([r.value for r in models.JobFailureTag])}" ) else: - rprint( - "[yellow]Could not determine Job ID to set failure reason " - "(printer has no active job info).[/yellow]" + common.output_message( + "Could not determine Job ID to set failure reason (printer has no active job info)." ) except Exception as e: - rprint(f"[red]Failed to set failure reason: {e}[/red]") - + common.output_message(f"Failed to set failure reason: {e}", error=True) else: - rprint(f"[red]Failed to send STOP_PRINT to {pid}[/red]") + common.output_message(f"Failed to send STOP_PRINT to {pid}", error=True) except Exception as e: - rprint(f"[red]Failed to send STOP_PRINT to {pid}: {e}[/red]") + common.output_message(f"Failed to send STOP_PRINT to {pid}: {e}", error=True) @printer_app.command(name="cancel-object") @@ -324,22 +316,18 @@ def printer_cancel_object( """Cancel a specific object during print.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return common.logger.debug("Command started", command="printer cancel-object", printer_id=resolved_id, object_id=object_id) client = common.get_client() try: if client.cancel_object(resolved_id, object_id): - rprint(f"[green]Successfully sent CANCEL_OBJECT for object {object_id} to {resolved_id}[/green]") + common.output_message(f"Successfully sent CANCEL_OBJECT for object {object_id} to {resolved_id}") else: - rprint("[red]Failed to send CANCEL_OBJECT command[/red]") + common.output_message("Failed to send CANCEL_OBJECT command", error=True) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @printer_app.command(name="move") @@ -354,22 +342,21 @@ def printer_move( """Move printer axis.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return + if not any([x, y, z, e]): + common.output_message("At least one axis (x, y, z, or e) must be specified", error=True) + common.logger.debug("Command started", command="printer move", printer_id=resolved_id) client = common.get_client() try: if client.move_axis(resolved_id, x=x, y=y, z=z, e=e, speed=speed): - rprint(f"[green]Successfully sent MOVE command to {resolved_id}[/green]") + common.output_message(f"Successfully sent MOVE command to {resolved_id}") else: - rprint("[red]Failed to send MOVE command[/red]") + common.output_message("Failed to send MOVE command", error=True) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @printer_app.command(name="flash") @@ -382,22 +369,18 @@ def printer_flash( """Flash firmware from a file on the printer's storage.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return common.logger.debug("Command started", command="printer flash", printer_id=resolved_id, file_path=file_path) client = common.get_client() try: if client.flash_firmware(resolved_id, file_path): - rprint(f"[green]Successfully sent FLASH command to {resolved_id}[/green]") + common.output_message(f"Successfully sent FLASH command to {resolved_id}") else: - rprint("[red]Failed to send FLASH command[/red]") + common.output_message("Failed to send FLASH command", error=True) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @printer_app.command(name="commands") @@ -407,11 +390,7 @@ def printer_commands( """List supported commands for a specific printer.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return common.logger.debug("Command started", command="printer commands", printer_id=resolved_id) @@ -421,14 +400,11 @@ def printer_commands( commands = client.get_supported_commands(resolved_id) # Deduplicate commands - # The API can return "duplicates" that differ only by internal fields like 'template'. - # We group by name and signature (arguments) to avoid showing redundant entries. unique_map: dict[str, list] = {} for cmd in commands: if cmd.command not in unique_map: unique_map[cmd.command] = [] - # Check for signature match is_duplicate = False current_args_json = json.dumps( [a.model_dump(include={"name", "type", "required"}) for a in cmd.args], sort_keys=True @@ -447,14 +423,8 @@ def printer_commands( unique_commands = [cmd for sublist in unique_map.values() for cmd in sublist] - table = Table(title=f"Supported Commands for {printer_id}") - table.add_column("Command", style="cyan") - table.add_column("Description", style="white") - table.add_column("Arguments", style="yellow") - table.add_column("Valid States", style="green") - + rows = [] for cmd in sorted(unique_commands, key=lambda x: x.command): - # Format arguments args_str = "" if cmd.args: arg_list = [] @@ -463,19 +433,24 @@ def printer_commands( arg_list.append(f"{arg.name}{req_mark}") args_str = ", ".join(arg_list) - # Format states states_str = ", ".join(cmd.executable_from_state) if cmd.executable_from_state else "ALL" if len(states_str) > 30: states_str = states_str[:27] + "..." - table.add_row(cmd.command, cmd.description or "", args_str, states_str) + rows.append([cmd.command, cmd.description or "", args_str, states_str]) + + common.output_table( + f"Supported Commands for {resolved_id}", + ["Command", "Description", "Arguments", "Valid States"], + rows, + column_styles=["cyan", "white", "yellow", "green"], + ) - common.console.print(table) if not commands: - rprint("[yellow]No supported commands found (or printer does not support command discovery).[/yellow]") + common.output_message("No supported commands found (or printer does not support command discovery).") except Exception as e: - rprint(f"[red]Error fetching commands:[/red] {e}") + common.output_message(f"Error fetching commands: {e}", error=True) @printer_app.command(name="command") @@ -506,11 +481,7 @@ def printer_execute_command( """ resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return common.logger.debug( @@ -527,7 +498,6 @@ def printer_execute_command( # 1. Parse Arguments final_args = {} - # Load from JSON if provided if args: try: json_parsed = json.loads(args) @@ -535,80 +505,58 @@ def printer_execute_command( raise ValueError("--args must be a JSON object") final_args.update(json_parsed) except json.JSONDecodeError as e: - rprint(f"[red]Invalid JSON in --args:[/red] {e}") + common.output_message(f"Invalid JSON in --args: {e}", error=True) return - # Load from kwargs (flags) - # cyclopts passes these as strings - # client.execute_printer_command does simple type checking: - # if arg_def.type == "integer" and not isinstance(val, int): raise... - # So we MUST cast here. - - # To cast correctly, we need the command definition! - - # Fetch definition first to know types supported = client.get_supported_commands(resolved_id) cmd_def = next((c for c in supported if c.command == command_name), None) if not cmd_def: - rprint(f"[red]Command '{command_name}' not supported by printer {resolved_id}.[/red]") - # Suggest close matches? + common.output_message(f"Command '{command_name}' not supported by printer {resolved_id}.", error=True) matches = fnmatch.filter([c.command for c in supported], f"*{command_name}*") if matches: - rprint(f"Did you mean: {', '.join(matches)}?") + common.output_message(f"Did you mean: {', '.join(matches)}?") return - # Merge kwargs into final_args, converting types for k, v in kwargs.items(): - # Match k to arg name (command args are snake_case usually) - # CLI flags are kebab-case but cyclopts normalizes them to snake_case - - # Find the argument definition arg_def = next((a for a in cmd_def.args if a.name == k), None) if arg_def: - # Cast based on type if arg_def.type == "integer": try: final_args[k] = int(v) except ValueError: - rprint(f"[red]Argument '{k}' must be an integer (got '{v}')[/red]") + common.output_message(f"Argument '{k}' must be an integer (got '{v}')", error=True) return elif arg_def.type == "number": try: final_args[k] = float(v) except ValueError: - rprint(f"[red]Argument '{k}' must be a number (got '{v}')[/red]") + common.output_message(f"Argument '{k}' must be a number (got '{v}')", error=True) return elif arg_def.type == "boolean": - # CLI flags for bools: usually presence means True - # cyclopts **kwargs treats flags with values if str(v).lower() in ("true", "1", "yes", "on"): final_args[k] = True elif str(v).lower() in ("false", "0", "no", "off"): final_args[k] = False else: - rprint(f"[red]Argument '{k}' must be a boolean (got '{v}')[/red]") + common.output_message(f"Argument '{k}' must be a boolean (got '{v}')", error=True) return else: final_args[k] = v else: - # Unknown argument - # Client doesn't strict check unknown extra args in 'execute_printer_command', - # it blindly passes everything to send_command after verifying *required*. - # But it DOES check known args for types. final_args[k] = v # 2. Execute success = client.execute_printer_command(resolved_id, command_name, final_args) if success: - rprint(f"[green]Successfully sent command '{command_name}' to {resolved_id}[/green]") + common.output_message(f"Successfully sent command '{command_name}' to {resolved_id}") else: - rprint(f"[red]Failed to send command '{command_name}'[/red]") + common.output_message(f"Failed to send command '{command_name}'", error=True) except ValueError as e: - rprint(f"[red]Validation Error:[/red] {e}") + common.output_message(f"Validation Error: {e}", error=True) except Exception as e: - rprint(f"[red]Error executing command:[/red] {e}") + common.output_message(f"Error executing command: {e}", error=True) @printer_app.command(name="storages") @@ -618,35 +566,32 @@ def printer_storages( """List storage devices attached to a printer.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() try: storages = client.get_printer_storages(resolved_id) - table = Table(title=f"Storages for {resolved_id}") - table.add_column("Name", style="cyan") - table.add_column("Type", style="green") - table.add_column("Mountpoint", style="magenta") - table.add_column("Free", style="yellow") - table.add_column("ReadOnly", style="red") - + rows = [] for s in storages: free_str = f"{s.free_space / 1024 / 1024 / 1024:.2f} GB" if s.free_space else "N/A" - table.add_row( - s.name, - s.type, - s.mountpoint or s.path, - free_str, - "Yes" if s.read_only else "No", + rows.append( + [ + s.name, + s.type, + s.mountpoint or s.path, + free_str, + "Yes" if s.read_only else "No", + ] ) - common.console.print(table) + common.output_table( + f"Storages for {resolved_id}", + ["Name", "Type", "Mountpoint", "Free", "ReadOnly"], + rows, + column_styles=["cyan", "green", "magenta", "yellow", "red"], + ) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @files_printer_app.command(name="list") @@ -656,32 +601,27 @@ def printer_files_list( """List files on the printer's storage.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() try: files = client.get_printer_files(resolved_id) - table = Table(title=f"Files on {resolved_id}") - table.add_column("Name", style="cyan") - table.add_column("Path", style="magenta") - table.add_column("Size", style="green") - table.add_column("Modified", style="yellow") - + rows = [] for f in files: size_str = f"{f.size / 1024 / 1024:.2f} MB" if f.size else "N/A" mtime_str = "N/A" if f.m_timestamp: mtime_str = datetime.datetime.fromtimestamp(f.m_timestamp).strftime("%Y-%m-%d %H:%M:%S") - - table.add_row(f.name, f.path, size_str, mtime_str) - common.console.print(table) + rows.append([f.name, f.path or "", size_str, mtime_str]) + common.output_table( + f"Files on {resolved_id}", + ["Name", "Path", "Size", "Modified"], + rows, + column_styles=["cyan", "magenta", "green", "yellow"], + ) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @files_printer_app.command(name="upload") @@ -693,25 +633,20 @@ def printer_files_upload( """Upload a file to a printer's storage.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() try: p = client.printers.get(resolved_id) teams = client.teams.list_teams() - # Find team by team_name target_team = next((t for t in teams if t.name == p.team_name), None) if not target_team and teams: target_team = teams[0] - rprint(f"[yellow]Could not resolve team ID for printer. Using first team: {target_team.name}[/yellow]") + common.output_message(f"Could not resolve team ID for printer. Using first team: {target_team.name}") if not target_team: - rprint("[red]Could not determine team for upload.[/red]") + common.output_message("Could not determine team for upload.", error=True) return from prusa.connect.client.cli.commands import file @@ -719,7 +654,7 @@ def printer_files_upload( file.file_upload(path=path, team_id=target_team.id, destination=destination) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @files_printer_app.command(name="download") @@ -731,11 +666,7 @@ def printer_files_download( """Download a file that belongs to a printer's team.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - rprint( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() @@ -747,7 +678,7 @@ def printer_files_download( target_team = teams[0] if not target_team: - rprint("[red]Could not determine team for download.[/red]") + common.output_message("Could not determine team for download.", error=True) return from prusa.connect.client.cli.commands import file @@ -755,7 +686,7 @@ def printer_files_download( file.file_download(file_hash=file_hash, team_id=target_team.id, output=output) except Exception as e: - rprint(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @printer_app.command(name="set-current") @@ -763,4 +694,4 @@ def set_current_printer(printer_id: typing.Annotated[str, cyclopts.Parameter(hel """Set the default printer UUID for future commands.""" config.settings.default_printer_id = printer_id config.save_json_config(config.settings) - rprint(f"[green]Successfully set default printer to {printer_id}[/green]") + common.output_message(f"Successfully set default printer to {printer_id}") diff --git a/src/prusa/connect/client/cli/commands/stats.py b/src/prusa/connect/client/cli/commands/stats.py index 28bbe4e..4a5c429 100644 --- a/src/prusa/connect/client/cli/commands/stats.py +++ b/src/prusa/connect/client/cli/commands/stats.py @@ -4,13 +4,18 @@ import typing import cyclopts -from rich.table import Table from prusa.connect.client.cli import common, config stats_app = cyclopts.App(name="stats", help="Printer statistics") logger = common.logger +_NO_PRINTER = ( + "No printer ID provided and no default configured.\n" + "Hint: Run 'prusactl printer list' to find a UUID, then " + "'prusactl printer set-current ' to set the default." +) + @stats_app.command(name="usage") def stats_usage( @@ -20,15 +25,12 @@ def stats_usage( datetime.date | None, cyclopts.Parameter(name=["--from", "-f"], help="Start date") ] = None, to_date: typing.Annotated[datetime.date | None, cyclopts.Parameter(name=["--to", "-t"], help="End date")] = None, + seconds: typing.Annotated[bool, cyclopts.Parameter(help="Output duration in seconds")] = False, ): """Show printer usage statistics (printing vs not printing).""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - common.console.print( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then\n" - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() @@ -39,16 +41,18 @@ def stats_usage( try: stats = client.get_printer_usage_stats(resolved_id, from_time=from_date, to_time=to_date) - table = Table(title=f"Usage Stats for {stats.printer_name} ({from_date} to {to_date})") - table.add_column("Type", style="cyan") - table.add_column("Value", style="magenta") - - for entry in stats.data: - table.add_row(entry.name, str(entry.value)) - - common.console.print(table) + rows = [ + [entry.name, str(entry.duration) if not seconds else str(entry.duration.total_seconds())] + for entry in stats.data + ] + common.output_table( + f"Usage Stats for {stats.printer_name} ({from_date} to {to_date})", + ["Type", "Duration"], + rows, + column_styles=["cyan", "magenta"], + ) except Exception as e: - common.console.print(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @stats_app.command(name="material") @@ -63,11 +67,7 @@ def stats_material( """Show material quantity statistics.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - common.console.print( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() @@ -79,22 +79,24 @@ def stats_material( try: stats = client.get_printer_material_stats(resolved_id, from_time=from_date, to_time=to_date) - table = Table(title=f"Material Stats for {stats.printer_name} ({from_date} to {to_date})") - table.add_column("Material", style="cyan") - table.add_column("Usage", style="magenta") - + rows = [] if not stats.data: - table.add_row("No data available", "") + rows.append(["No data available", ""]) else: for entry in stats.data: if isinstance(entry, dict): - table.add_row(entry.get("name", "Unknown"), str(entry.get("value", "N/A"))) + rows.append([entry.get("name", "Unknown"), str(entry.get("value", "N/A"))]) else: - table.add_row("Raw Data", str(entry)) + rows.append(["Raw Data", str(entry)]) - common.console.print(table) + common.output_table( + f"Material Stats for {stats.printer_name} ({from_date} to {to_date})", + ["Material", "Usage"], + rows, + column_styles=["cyan", "magenta"], + ) except Exception as e: - common.console.print(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @stats_app.command(name="jobs") @@ -109,11 +111,7 @@ def stats_jobs( """Show job success statistics.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - common.console.print( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() @@ -129,20 +127,20 @@ def stats_jobs( stats.series.sort(key=lambda x: x.status) logger.debug("Job Stats", data=stats) - table = Table(title=f"Job Success Stats for {stats.printer_name} ({from_date} to {to_date})") - table.add_column("Status", style="cyan") - - for date in stats.date_axis: - table.add_column(date, style="magenta") - + columns = ["Status", *list(stats.date_axis)] + rows = [] for series in stats.series: - row = [series.status.name] - row.extend(str(v) for v in series.data) - table.add_row(*row) - - common.console.print(table) + row = [series.status.name] + [str(v) for v in series.data] + rows.append(row) + + common.output_table( + f"Job Success Stats for {stats.printer_name} ({from_date} to {to_date})", + columns, + rows, + column_styles=["cyan"] + ["magenta"] * len(stats.date_axis), + ) except Exception as e: - common.console.print(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) @stats_app.command(name="planned") @@ -157,11 +155,7 @@ def stats_planned( """Show planned tasks statistics.""" resolved_id = printer_id or config.settings.default_printer_id if not resolved_id: - common.console.print( - "[red]No printer ID provided and no default configured.[/red]\n" - "[dim]Hint: Run 'prusactl printer list' to find a UUID, then " - "'prusactl printer set-current ' to set the default.[/dim]" - ) + common.output_message(_NO_PRINTER, error=True) return client = common.get_client() @@ -172,16 +166,19 @@ def stats_planned( try: stats = client.get_printer_planned_tasks_stats(resolved_id, from_time=from_date, to_time=to_date) - table = Table(title=f"Planned Tasks for {stats.series.printer_name} ({from_date} to {to_date})") - table.add_column("Hour (UTC)", style="cyan") - table.add_column("Count", style="magenta") + rows = [] if stats.series and stats.series.data: for hour, count in stats.series.data: - table.add_row(f"{hour:02d}:00", str(count)) + rows.append([f"{hour:02d}:00", str(count)]) else: - table.add_row("No data available", "") + rows.append(["No data available", ""]) - common.console.print(table) + common.output_table( + f"Planned Tasks for {stats.series.printer_name} ({from_date} to {to_date})", + ["Hour (UTC)", "Count"], + rows, + column_styles=["cyan", "magenta"], + ) except Exception as e: - common.console.print(f"[red]Error:[/red] {e}") + common.output_message(f"Error: {e}", error=True) diff --git a/src/prusa/connect/client/cli/commands/team.py b/src/prusa/connect/client/cli/commands/team.py index 7426405..bdbf8a0 100644 --- a/src/prusa/connect/client/cli/commands/team.py +++ b/src/prusa/connect/client/cli/commands/team.py @@ -1,11 +1,10 @@ """Team commands.""" +import json import sys import typing import cyclopts -from rich import print as rprint -from rich.table import Table from prusa.connect.client.cli import common, config from prusa.connect.client.cli.commands.job import job_list @@ -19,20 +18,13 @@ def list_teams(): client = common.get_client() teams = client.teams.list_teams() - table = Table(title="Teams") - table.add_column("ID", style="cyan") - table.add_column("Name", style="green") - table.add_column("Role", style="magenta") - table.add_column("Organization ID", style="blue") - - for team in teams: - table.add_row( - str(team.id), - team.name, - str(team.role or "N/A"), - str(team.organization_id or "N/A"), - ) - rprint(table) + rows = [[str(team.id), team.name, str(team.role or "N/A"), str(team.organization_id or "N/A")] for team in teams] + common.output_table( + "Teams", + ["ID", "Name", "Role", "Organization ID"], + rows, + column_styles=["cyan", "green", "magenta", "blue"], + ) @team_app.command(name="show") @@ -45,41 +37,39 @@ def show_team( """Show details for a specific team.""" team_id_to_use = team_id or config.settings.default_team_id if team_id_to_use is None: - rprint("[red]Error: Team ID not provided and no default is set.[/red]") + common.output_message("Error: Team ID not provided and no default is set.", error=True) sys.exit(1) client = common.get_client() try: team = client.teams.get(team_id_to_use) except Exception as e: - rprint(f"[red]Error fetching team {team_id_to_use}: {e}[/red]") + common.output_message(f"Error fetching team {team_id_to_use}: {e}", error=True) sys.exit(1) - table = Table(title=f"Team Details: {team.name}") - table.add_column("Property", style="cyan") - table.add_column("Value", style="green") - - table.add_row("ID", str(team.id)) - table.add_row("Name", team.name) - table.add_row("Role", str(team.role or "N/A")) + rows = [ + ["ID", str(team.id)], + ["Name", team.name], + ["Role", str(team.role or "N/A")], + ] if team.description: - table.add_row("Description", team.description) + rows.append(["Description", team.description]) if team.capacity is not None: - table.add_row("Capacity", str(team.capacity)) + rows.append(["Capacity", str(team.capacity)]) if team.organization_id: - table.add_row("Organization ID", str(team.organization_id)) + rows.append(["Organization ID", str(team.organization_id)]) if team.user_count is not None: - table.add_row("User Count", str(team.user_count)) + rows.append(["User Count", str(team.user_count)]) - rprint(table) + common.output_table( + f"Team Details: {team.name}", + ["Property", "Value"], + rows, + column_styles=["cyan", "green"], + ) if getattr(team, "users", None): - users_table = Table(title="Team Users") - users_table.add_column("ID", style="cyan") - users_table.add_column("Name", style="green") - users_table.add_column("Username", style="magenta") - users_table.add_column("Rights", style="yellow") - + user_rows = [] for u in team.users: name_parts = [p for p in [u.first_name, u.last_name] if p] name = " ".join(name_parts) if name_parts else "N/A" @@ -90,15 +80,17 @@ def show_team( rights.append("RW") if u.rights_use: rights.append("USE") + user_rows.append([str(u.id), name, u.public_name or "N/A", ", ".join(rights) if rights else "NONE"]) - users_table.add_row(str(u.id), name, u.public_name or "N/A", ", ".join(rights) if rights else "NONE") - rprint(users_table) + common.output_table( + "Team Users", + ["ID", "Name", "Username", "Rights"], + user_rows, + column_styles=["cyan", "green", "magenta", "yellow"], + ) if detailed: - import json - - rprint("\n[bold]Detailed Information:[/bold]") - detail_table = Table(show_header=False, box=None) + detail_rows = [] for k, v in team.model_dump(mode="json").items(): if v is not None and k not in [ "id", @@ -111,8 +103,15 @@ def show_team( "users", ]: val_str = json.dumps(v) if isinstance(v, (dict, list)) else str(v) - detail_table.add_row(f"[cyan]{k}[/cyan]:", val_str) - common.console.print(detail_table) + detail_rows.append([k, val_str]) + + if detail_rows: + common.output_table( + "Detailed Information", + ["Field", "Value"], + detail_rows, + column_styles=["cyan", None], + ) @team_app.command(name="add-user") @@ -126,27 +125,25 @@ def add_team_user( """Invite a user to a team.""" team_id_to_use = team_id or config.settings.default_team_id if team_id_to_use is None: - rprint("[red]Error: Team ID not provided and no default is set.[/red]") + common.output_message("Error: Team ID not provided and no default is set.", error=True) sys.exit(1) client = common.get_client() try: if client.add_team_user(team_id_to_use, email, rights_ro, rights_use, rights_rw): - rprint(f"[green]Successfully sent invitation to {email}[/green]") + common.output_message(f"Successfully sent invitation to {email}") except Exception as e: from prusa.connect.client import exceptions if isinstance(e, exceptions.PrusaApiError): - import json - try: err_data = json.loads(e.response_body) msg = err_data.get("message", e.response_body) - rprint(f"[red]Failed to add user:[/red] {msg}") + common.output_message(f"Failed to add user: {msg}", error=True) except json.JSONDecodeError: - rprint(f"[red]Failed to add user:[/red] {e.response_body}") + common.output_message(f"Failed to add user: {e.response_body}", error=True) else: - rprint(f"[red]Failed to add user: {e}[/red]") + common.output_message(f"Failed to add user: {e}", error=True) @team_app.command(name="set-current") @@ -156,7 +153,7 @@ def set_current_team( """Set the default team ID for future commands.""" config.settings.default_team_id = team_id config.save_json_config(config.settings) - rprint(f"[green]Successfully set default team to {team_id}[/green]") + common.output_message(f"Successfully set default team to {team_id}") def teams_alias(): @@ -174,7 +171,7 @@ def team_jobs_alias( """List jobs (alias for 'job list').""" team_id_to_use = team or config.settings.default_team_id if team_id_to_use is None: - rprint("[red]Error: Team ID not provided and no default is set.[/red]") + common.output_message("Error: Team ID not provided and no default is set.", error=True) sys.exit(1) job_list(team=team_id_to_use, printer=printer, state=state, limit=limit) diff --git a/src/prusa/connect/client/cli/common.py b/src/prusa/connect/client/cli/common.py index 39685f1..3fd9577 100644 --- a/src/prusa/connect/client/cli/common.py +++ b/src/prusa/connect/client/cli/common.py @@ -1,5 +1,7 @@ """Shared CLI helpers and configuration.""" +import collections.abc +import json as _json import logging import pathlib import sys @@ -9,7 +11,7 @@ import platformdirs import structlog from rich import console as rich_console -from rich import print as rprint +from rich.text import Text from prusa.connect.client import auth, exceptions, sdk from prusa.connect.client import consts as sdk_consts @@ -21,8 +23,110 @@ # Setup better_exceptions.hook() console = rich_console.Console() +err_console = rich_console.Console(stderr=True) logger = structlog.get_logger(sdk_consts.APP_NAME) +# -- Output format ---------------------------------------------------------- + +_output_format: config.OutputFormat | None = None # None means "resolve lazily from config/TTY" + + +def set_output_format(fmt: str | None) -> None: + """Set the output format (called from --format CLI flag). + + Calls `sys.exit` if an invalid format is specified. + """ + global _output_format + try: + _output_format = config.OutputFormat(fmt) if fmt is not None else None + except ValueError: + output_message( + ( + f"[bold][red]Error[/red][/bold]: `{fmt}` is not a valid output " + f"format. Valid formats are: {', '.join(config.OutputFormat.__members__.values())}" + ), + error=True, + ) + sys.exit(1) + + +def get_output_format() -> config.OutputFormat: + """Resolve the active output format: CLI flag > config > TTY auto-detect.""" + if _output_format is not None: + return _output_format + cfg_fmt = getattr(config.settings, "output_format", None) + if cfg_fmt: + return cfg_fmt + return config.OutputFormat.RICH if sys.stdout.isatty() else config.OutputFormat.PLAIN + + +def _strip_markup(text: str) -> str: + """Remove Rich markup tags from a string.""" + return Text.from_markup(str(text)).plain + + +def output_message(msg: str, *, error: bool = False) -> None: + """Print a status/error message respecting the current output format. + + - rich: renders markup with color to stdout (or stderr for errors) + - plain: strips markup, writes to stdout (errors to stderr) + - json: strips markup, always writes to stderr (stdout reserved for JSON) + """ + fmt = get_output_format() + if fmt in ("plain", "json"): + plain = _strip_markup(msg) + to_stderr = error or fmt == "json" + print(plain, file=sys.stderr if to_stderr else sys.stdout) + else: + target = err_console if error else console + target.print(msg) + + +def output_table( + title: str, + columns: list[str], + rows: list[list[str]], + *, + column_styles: collections.abc.Sequence[str | None] | None = None, + sections_before: set[int] | None = None, +) -> None: + """Print tabular data respecting the current output format. + + Args: + title: Table title (used as rich title; as ``# title`` comment in plain). + columns: Column header names. + rows: Row data as lists of strings (may contain Rich markup; stripped in + plain/json modes). + column_styles: Optional per-column Rich style names (ignored in plain/json). + sections_before: Set of row indices before which ``table.add_section()`` + is called (rich only; ignored in plain/json). + """ + from rich.table import Table as _RichTable + + fmt = get_output_format() + + if fmt == "json": + keys = [c.lower().replace(" ", "_").replace("(", "").replace(")", "").strip("_") for c in columns] + data = [dict(zip(keys, [_strip_markup(c) for c in row], strict=False)) for row in rows] + print(_json.dumps(data)) + elif fmt == "plain": + print(f"# {title}") + print("\t".join(columns)) + for row in rows: + print("\t".join(_strip_markup(c) for c in row)) + else: + table = _RichTable(title=title) + styles = column_styles or [] + for i, col in enumerate(columns): + style = styles[i] if i < len(styles) else None + table.add_column(col, style=style) + for i, row in enumerate(rows): + if sections_before and i in sections_before: + table.add_section() + table.add_row(*[str(c) for c in row]) + console.print(table) + + _LOGGING_INITIALIZED = False @@ -102,8 +206,8 @@ def get_client(require_auth: bool = True) -> sdk.PrusaConnectClient: creds = None if (creds is None or not creds.valid) and require_auth: - rprint("[red]Authentication required.[/red]") - rprint("Please run [bold]prusactl auth login[/bold] to authenticate.") + output_message("Authentication required.", error=True) + output_message("Please run 'prusactl auth login' to authenticate.", error=True) sys.exit(1) cache_dir = pathlib.Path(platformdirs.user_cache_dir(sdk_consts.APP_NAME, sdk_consts.APP_AUTHOR)) diff --git a/src/prusa/connect/client/cli/config.py b/src/prusa/connect/client/cli/config.py index 95f51c0..b9f1eb7 100644 --- a/src/prusa/connect/client/cli/config.py +++ b/src/prusa/connect/client/cli/config.py @@ -1,5 +1,6 @@ """Configuration handling for the CLI.""" +import enum import json import pathlib import typing @@ -39,6 +40,14 @@ def load_json_config() -> dict[str, typing.Any]: return {} +class OutputFormat(enum.StrEnum): + """Enum of available console out formats.""" + + RICH = "rich" + PLAIN = "plain" + JSON = "json" + + class Settings(pydantic_settings.BaseSettings): """Application-wide settings loaded from config.json, .env or environment variables.""" @@ -52,6 +61,14 @@ class Settings(pydantic_settings.BaseSettings): tokens_file: pathlib.Path = pydantic.Field(default_factory=auth.get_default_token_path) cache_ttl_hours: int = consts.DEFAULT_CACHE_TTL_HOURS + # Output format: "rich" (colored tables), "plain" (tab-separated, no markup), + # or "json" (JSON arrays to stdout). None means auto-detect from TTY. + # Override via env var PRUSACTL_OUTPUT_FORMAT or config.json key "output_format". + output_format: OutputFormat | None = pydantic.Field( + default=None, + validation_alias=pydantic.AliasChoices("PRUSACTL_OUTPUT_FORMAT", "output_format"), + ) + model_config = pydantic_settings.SettingsConfigDict(env_file=".env", extra="ignore") @classmethod diff --git a/src/prusa/connect/client/cli/main.py b/src/prusa/connect/client/cli/main.py index 7bf1875..41ce7ea 100644 --- a/src/prusa/connect/client/cli/main.py +++ b/src/prusa/connect/client/cli/main.py @@ -29,7 +29,7 @@ # Register Aliases and Commands app.command(printer.printers_alias, name="printers") -app.command(camera.camera_alias, name="cameras") +app.command(camera.cameras_alias, name="cameras") app.command(job.jobs_alias, name="jobs") app.command(file.files_alias, name="files") app.command(team.teams_alias, name="teams") @@ -44,11 +44,21 @@ def entry_point( bool, cyclopts.Parameter(name=["--verbose", "-v"], help="Enable verbose logging") ] = False, debug: typing.Annotated[bool, cyclopts.Parameter(name=["--debug"], help="Enable debug logging")] = False, + output_format: typing.Annotated[ + str | None, + cyclopts.Parameter( + name=["--format"], + help="Output format: rich (default on TTY), plain (tab-separated), json", + ), + ] = None, ): """Main entry point handling global flags.""" # Configure logging common.configure_logging(verbose, debug) + if output_format is not None: + common.set_output_format(output_format) + if tokens is None: tokens = [] # Let cyclopts handle the full command parsing (subcommands, help, etc) diff --git a/src/prusa/connect/client/models/stats.py b/src/prusa/connect/client/models/stats.py index 6d107ef..c53e05c 100644 --- a/src/prusa/connect/client/models/stats.py +++ b/src/prusa/connect/client/models/stats.py @@ -61,7 +61,7 @@ class PrintingNotPrintingEntry(WarnExtraFieldsModel): """Represents a single entry in printing vs not printing stats.""" name: str - value: int + duration: datetime.timedelta = pydantic.Field(..., alias="value") class PrintingNotPrinting(StatsModel): diff --git a/tests/unit_tests/test_cli_output.py b/tests/unit_tests/test_cli_output.py new file mode 100644 index 0000000..2ebcf1e --- /dev/null +++ b/tests/unit_tests/test_cli_output.py @@ -0,0 +1,109 @@ +import contextlib +import json +import sys +from unittest.mock import MagicMock, patch + +import pytest + +from prusa.connect.client.cli import common, config + + +@pytest.fixture(autouse=True) +def reset_output_format(): + """Reset the global output format before and after each test.""" + common._output_format = None + yield + common._output_format = None + + +def test_set_output_format_valid(): + common.set_output_format("json") + assert common.get_output_format() == config.OutputFormat.JSON + + common.set_output_format("plain") + assert common.get_output_format() == config.OutputFormat.PLAIN + + common.set_output_format("rich") + assert common.get_output_format() == config.OutputFormat.RICH + + +def test_set_output_format_invalid(): + with pytest.raises(SystemExit) as excinfo: + common.set_output_format("invalid") + assert excinfo.value.code == 1 + + +def test_get_output_format_default_tty(monkeypatch): + # Mock sys.stdout.isatty to return True + monkeypatch.setattr(sys.stdout, "isatty", lambda: True) + assert common.get_output_format() == config.OutputFormat.RICH + + +def test_get_output_format_default_no_tty(monkeypatch): + # Mock sys.stdout.isatty to return False + monkeypatch.setattr(sys.stdout, "isatty", lambda: False) + assert common.get_output_format() == config.OutputFormat.PLAIN + + +def test_get_output_format_from_config(): + with patch("prusa.connect.client.cli.config.settings") as mock_settings: + mock_settings.output_format = config.OutputFormat.JSON + assert common.get_output_format() == config.OutputFormat.JSON + + +def test_output_message_rich(capsys): + common.set_output_format("rich") + # We can't easily test rich's actual colored output here because it depends on terminal + # but we can check if it calls the console. + with patch("prusa.connect.client.cli.common.console") as mock_console: + common.output_message("Hello [bold]World[/bold]") + mock_console.print.assert_called_once_with("Hello [bold]World[/bold]") + + +def test_output_message_plain(capsys): + common.set_output_format("plain") + common.output_message("Hello [bold]World[/bold]") + captured = capsys.readouterr() + assert captured.out == "Hello World\n" + + +def test_output_message_json(capsys): + common.set_output_format("json") + # JSON format should send messages to stderr + common.output_message("Hello [bold]World[/bold]") + captured = capsys.readouterr() + assert "Hello World" in captured.err + + +def test_output_table_plain(capsys): + common.set_output_format("plain") + common.output_table("My Table", ["Col1", "Col2"], [["R1C1", "R1C2"], ["R2C1", "R2C2"]]) + captured = capsys.readouterr() + expected = "# My Table\nCol1\tCol2\nR1C1\tR1C2\nR2C1\tR2C2\n" + assert captured.out == expected + + +def test_output_table_json(capsys): + common.set_output_format("json") + common.output_table("My Table", ["Col 1", "Col 2"], [["R1C1", "R1C2"]]) + captured = capsys.readouterr() + data = json.loads(captured.out) + assert data == [{"col_1": "R1C1", "col_2": "R1C2"}] + + +def test_cli_format_flag(): + from prusa.connect.client.cli.main import app + + # Use a command that output something, like 'printer list' + # We need to mock the client to avoid network calls + with patch("prusa.connect.client.cli.commands.printer.common.get_client") as mock_get_client: + client = MagicMock() + client.printers.list_printers.return_value = [] + mock_get_client.return_value = client + + # Test with --format json + with patch("prusa.connect.client.cli.common.set_output_format") as mock_set, contextlib.suppress(SystemExit): + # Use app.meta to handle global flags + app.meta(["--format", "json", "printer", "list"]) + + mock_set.assert_called_with("json") diff --git a/tests/unit_tests/test_stats.py b/tests/unit_tests/test_stats.py index 3427c83..bc6ea20 100644 --- a/tests/unit_tests/test_stats.py +++ b/tests/unit_tests/test_stats.py @@ -39,7 +39,7 @@ def test_get_printer_usage_stats(client): assert stats.printer_name == "printer-name" assert len(stats.data) == 2 assert stats.data[0].name == "printing" - assert stats.data[0].value == 10 + assert stats.data[0].duration.total_seconds() == 10.0 assert isinstance(stats.from_time, datetime.date)