nwp500.cli package

Submodules

nwp500.cli.commands module

Command registry for NWP500 CLI.

class nwp500.cli.commands.CliCommand(name: str, help: str, callback: Callable[[...], Any], args: list[str], options: list[str], examples: list[str])

Bases: object

Definition of a CLI command.

args: list[str]

callback: Callable[[...], Any]

examples: list[str]

help: str

name: str

options: list[str]

nwp500.cli.commands.get_command(name: str) → CliCommand | None

Lookup command by name.

nwp500.cli.commands.list_commands() → list[CliCommand]

Get all available commands.

nwp500.cli.handlers module

Command handlers for CLI operations.

async nwp500.cli.handlers.get_controller_serial_number(mqtt: NavienMqttClient, device: Device, timeout: float = 10.0) → str | None

Retrieve controller serial number from device.

async nwp500.cli.handlers.handle_add_reservation_request(mqtt: NavienMqttClient, device: Device, enabled: bool, days: str, hour: int, minute: int, mode: int, temperature: float) → None

Add a single reservation to the existing schedule.

async nwp500.cli.handlers.handle_configure_reservation_water_program_request(mqtt: NavienMqttClient, device: Device) → None

Configure water program.

async nwp500.cli.handlers.handle_delete_reservation_request(mqtt: NavienMqttClient, device: Device, index: int) → None

Delete a single reservation by 1-based index.

async nwp500.cli.handlers.handle_device_info_request(mqtt: NavienMqttClient, device: Device, raw: bool = False) → None

Request comprehensive device information.

async nwp500.cli.handlers.handle_disable_anti_legionella_request(mqtt: NavienMqttClient, device: Device) → None

Disable Anti-Legionella disinfection cycle.

async nwp500.cli.handlers.handle_disable_demand_response_request(mqtt: NavienMqttClient, device: Device) → None

Disable demand response.

async nwp500.cli.handlers.handle_enable_anti_legionella_request(mqtt: NavienMqttClient, device: Device, period_days: int) → None

Enable Anti-Legionella disinfection cycle.

async nwp500.cli.handlers.handle_enable_demand_response_request(mqtt: NavienMqttClient, device: Device) → None

Enable demand response.

async nwp500.cli.handlers.handle_get_anti_legionella_status_request(mqtt: NavienMqttClient, device: Device) → None

Display Anti-Legionella status from device status.

async nwp500.cli.handlers.handle_get_controller_serial_request(mqtt: NavienMqttClient, device: Device) → None

Request and display just the controller serial number.

async nwp500.cli.handlers.handle_get_device_info_rest(api_client: NavienAPIClient, device: Device, raw: bool = False) → None

Get device info from REST API (minimal DeviceInfo fields).

async nwp500.cli.handlers.handle_get_energy_request(mqtt: NavienMqttClient, device: Device, year: int, months: list[int]) → None

Request energy usage data.

If a single month is provided, shows daily breakdown. If multiple months are provided, shows monthly summary.

async nwp500.cli.handlers.handle_get_reservations_request(mqtt: NavienMqttClient, device: Device, output_json: bool = False) → None

Request current reservation schedule.

async nwp500.cli.handlers.handle_get_tou_request(mqtt: NavienMqttClient, device: Device, api_client: Any, *, output_json: bool = False) → None

Request Time-of-Use settings from REST API.

async nwp500.cli.handlers.handle_power_request(mqtt: NavienMqttClient, device: Device, power_on: bool) → None

Set device power state.

async nwp500.cli.handlers.handle_reset_air_filter_request(mqtt: NavienMqttClient, device: Device) → None

Reset air filter timer.

async nwp500.cli.handlers.handle_set_anti_legionella_period_request(mqtt: NavienMqttClient, device: Device, period_days: int) → None

Set Anti-Legionella cycle period without changing enabled state.

async nwp500.cli.handlers.handle_set_dhw_temp_request(mqtt: NavienMqttClient, device: Device, temperature: float) → None

Set DHW target temperature.

async nwp500.cli.handlers.handle_set_mode_request(mqtt: NavienMqttClient, device: Device, mode_name: str) → None

Set device operation mode.

async nwp500.cli.handlers.handle_set_recirculation_mode_request(mqtt: NavienMqttClient, device: Device, mode: int) → None

Set recirculation pump mode.

async nwp500.cli.handlers.handle_set_tou_enabled_request(mqtt: NavienMqttClient, device: Device, enabled: bool) → None

Enable or disable Time-of-Use.

async nwp500.cli.handlers.handle_set_vacation_days_request(mqtt: NavienMqttClient, device: Device, days: int) → None

Set vacation mode duration.

async nwp500.cli.handlers.handle_status_request(mqtt: NavienMqttClient, device: Device, raw: bool = False) → None

Request device status and print it.

async nwp500.cli.handlers.handle_tou_apply_request(mqtt: NavienMqttClient, device: Device, api_client: NavienAPIClient, zip_code: str, plan_name: str, utility: str | None = None, enable: bool = False) → None

Apply a TOU rate plan to the water heater.

async nwp500.cli.handlers.handle_tou_plan_request(api_client: NavienAPIClient, zip_code: str, plan_name: str, utility: str | None = None, *, output_json: bool = False) → None

View a converted rate plan’s details.

async nwp500.cli.handlers.handle_tou_rates_request(zip_code: str, utility: str | None = None) → None

List utilities and rate plans for a zip code.

async nwp500.cli.handlers.handle_trigger_recirculation_hot_button_request(mqtt: NavienMqttClient, device: Device) → None

Trigger hot button.

async nwp500.cli.handlers.handle_update_reservation_request(mqtt: NavienMqttClient, device: Device, index: int, *, enabled: bool | None = None, days: str | None = None, hour: int | None = None, minute: int | None = None, mode: int | None = None, temperature: float | None = None) → None

Update a single reservation by 1-based index.

Only the provided fields are modified; others are preserved.

async nwp500.cli.handlers.handle_update_reservations_request(mqtt: NavienMqttClient, device: Device, reservations_json: str, enabled: bool) → None

Update reservation schedule.

nwp500.cli.monitoring module

Monitoring and periodic status polling.

async nwp500.cli.monitoring.handle_monitoring(mqtt: NavienMqttClient, device: Device, output_file: str) → None

Start periodic monitoring and write status to CSV.

Parameters:

• mqtt – MQTT client instance

• device – Device to monitor

• output_file – Path to output CSV file

This function runs indefinitely, polling the device every 30 seconds and writing status updates to a CSV file.

nwp500.cli.output_formatters module

Output formatting utilities for CLI (CSV, JSON).

nwp500.cli.output_formatters.format_daily_energy_usage(energy_response: Any, year: int, month: int) → str

Format daily energy usage for a specific month as a human-readable table.

Parameters:

• energy_response – EnergyUsageResponse object

• year – Year to filter for (e.g., 2025)

• month – Month to filter for (1-12)

Returns:

Formatted string with daily energy usage data in tabular form

nwp500.cli.output_formatters.format_energy_usage(energy_response: Any) → str

Format energy usage response as a human-readable table.

Parameters:

energy_response – EnergyUsageResponse object

Returns:

Formatted string with energy usage data in tabular form

nwp500.cli.output_formatters.format_json_output(data: Any, indent: int = 2) → str

Format data as JSON string with custom serialization.

Parameters:

• data – Data to format

• indent – Number of spaces for indentation (default: 2)

Returns:

JSON-formatted string

nwp500.cli.output_formatters.print_daily_energy_usage(energy_response: Any, year: int, month: int) → None

Print daily energy usage data in human-readable tabular format.

Uses Rich formatting when available, falls back to plain text otherwise.

Parameters:

• energy_response – EnergyUsageResponse object

• year – Year to filter for (e.g., 2025)

• month – Month to filter for (1-12)

nwp500.cli.output_formatters.print_device_info(device_feature: Any) → None

Print device information with aligned columns and dynamic width calculation.

Parameters:

device_feature – DeviceFeature object

nwp500.cli.output_formatters.print_device_status(device_status: Any) → None

Print device status with aligned columns and dynamic width calculation.

Units are automatically extracted from the DeviceStatus model metadata.

Parameters:

device_status – DeviceStatus object

nwp500.cli.output_formatters.print_energy_usage(energy_response: Any) → None

Print energy usage data in human-readable tabular format.

Uses Rich formatting when available, falls back to plain text otherwise.

Parameters:

energy_response – EnergyUsageResponse object

nwp500.cli.output_formatters.print_json(data: Any, indent: int = 2) → None

Print data as formatted JSON with optional syntax highlighting.

Uses Rich highlighting when available, falls back to plain JSON otherwise.

Parameters:

• data – Data to print

• indent – Number of spaces for indentation (default: 2)

nwp500.cli.output_formatters.write_status_to_csv(file_path: str, status: DeviceStatus) → None

Append device status to a CSV file.

Parameters:

• file_path – Path to the CSV file

• status – DeviceStatus object to write

nwp500.cli.rich_output module

Rich-enhanced output formatting with graceful fallback.

class nwp500.cli.rich_output.OutputFormatter

Bases: object

Unified output formatter with Rich enhancement support.

Automatically detects Rich availability and routes output to the appropriate formatter. Falls back to plain text when Rich is unavailable or explicitly disabled.

print_daily_energy_table(days: list[dict[str, Any]], year: int, month: int) → None

Print daily energy usage data as a formatted table.

Parameters:

• days – List of daily energy data dictionaries

• year – Year for the data

• month – Month for the data

print_device_list(devices: list[dict[str, Any]]) → None

Print list of devices with status indicators.

Parameters:

devices – List of device dictionaries with status info

print_device_tree(device_name: str, device_info: dict[str, Any]) → None

Print device information as a tree structure.

Parameters:

• device_name – Name of the device

• device_info – Dictionary of device information

print_energy_table(months: list[dict[str, Any]]) → None

Print energy usage data as a formatted table.

Parameters:

months – List of monthly energy data dictionaries

print_error(message: str, title: str = 'Error', details: list[str] | None = None) → None

Print an error message.

Parameters:

• message – Main error message

• title – Panel title

• details – Optional list of detail lines

print_info(message: str) → None

Print an info message.

Parameters:

message – Info message to display

print_json_highlighted(data: Any) → None

Print JSON with syntax highlighting.

Parameters:

data – Data to print as JSON

print_markdown_report(markdown_content: str) → None

Print markdown-formatted content.

Parameters:

markdown_content – Markdown formatted string

print_reservations_table(reservations: list[dict[str, Any]], enabled: bool = False) → None

Print reservations as a formatted table.

Parameters:

• reservations – List of reservation dictionaries

• enabled – Whether reservations are enabled globally

print_status_table(items: list[tuple[str, str, str]]) → None

Print status items as a formatted table.

Parameters:

items – List of (category, label, value) tuples

print_success(message: str) → None

Print a success message.

Parameters:

message – Success message to display

print_tou_schedule(name: str, utility: str, zip_code: int, schedules: Any, decode_season: Any, decode_week: Any, decode_price_fn: Any) → None

Print TOU schedule as a human-readable table.

Parameters:

• name – Rate plan name

• utility – Utility company name

• zip_code – Service ZIP code

• schedules – List of TOUSchedule objects

• decode_season – Function to decode season bitfield

• decode_week – Function to decode week bitfield

• decode_price_fn – Function to decode price values

nwp500.cli.rich_output.get_formatter() → OutputFormatter

Get the global formatter instance.

Returns:

OutputFormatter instance with Rich support if available.

nwp500.cli.token_storage module

Token storage and management for CLI authentication.

nwp500.cli.token_storage.load_tokens() → tuple[AuthTokens | None, str | None]

Load authentication tokens and user email from a file.

Returns:

Tuple of (AuthTokens, email) or (None, None) if tokens cannot be loaded

nwp500.cli.token_storage.save_tokens(tokens: AuthTokens, email: str) → None

Save authentication tokens and user email to a file.

Parameters:

• tokens – AuthTokens object containing credentials

• email – User email address

Module contents

CLI package for nwp500-python.

nwp500.cli.format_json_output(data: Any, indent: int = 2) → str

Format data as JSON string with custom serialization.

Parameters:

• data – Data to format

• indent – Number of spaces for indentation (default: 2)

Returns:

JSON-formatted string

async nwp500.cli.handle_device_info_request(mqtt: NavienMqttClient, device: Device, raw: bool = False) → None

Request comprehensive device information.

async nwp500.cli.handle_get_controller_serial_request(mqtt: NavienMqttClient, device: Device) → None

Request and display just the controller serial number.

async nwp500.cli.handle_get_device_info_rest(api_client: NavienAPIClient, device: Device, raw: bool = False) → None

Get device info from REST API (minimal DeviceInfo fields).

async nwp500.cli.handle_get_energy_request(mqtt: NavienMqttClient, device: Device, year: int, months: list[int]) → None

Request energy usage data.

If a single month is provided, shows daily breakdown. If multiple months are provided, shows monthly summary.

async nwp500.cli.handle_get_reservations_request(mqtt: NavienMqttClient, device: Device, output_json: bool = False) → None

Request current reservation schedule.

async nwp500.cli.handle_get_tou_request(mqtt: NavienMqttClient, device: Device, api_client: Any, *, output_json: bool = False) → None

Request Time-of-Use settings from REST API.

async nwp500.cli.handle_monitoring(mqtt: NavienMqttClient, device: Device, output_file: str) → None

Start periodic monitoring and write status to CSV.

Parameters:

• mqtt – MQTT client instance

• device – Device to monitor

• output_file – Path to output CSV file

This function runs indefinitely, polling the device every 30 seconds and writing status updates to a CSV file.

async nwp500.cli.handle_power_request(mqtt: NavienMqttClient, device: Device, power_on: bool) → None

Set device power state.

async nwp500.cli.handle_set_dhw_temp_request(mqtt: NavienMqttClient, device: Device, temperature: float) → None

Set DHW target temperature.

async nwp500.cli.handle_set_mode_request(mqtt: NavienMqttClient, device: Device, mode_name: str) → None

Set device operation mode.

async nwp500.cli.handle_set_tou_enabled_request(mqtt: NavienMqttClient, device: Device, enabled: bool) → None

Enable or disable Time-of-Use.

async nwp500.cli.handle_status_request(mqtt: NavienMqttClient, device: Device, raw: bool = False) → None

Request device status and print it.

async nwp500.cli.handle_tou_apply_request(mqtt: NavienMqttClient, device: Device, api_client: NavienAPIClient, zip_code: str, plan_name: str, utility: str | None = None, enable: bool = False) → None

Apply a TOU rate plan to the water heater.

async nwp500.cli.handle_tou_plan_request(api_client: NavienAPIClient, zip_code: str, plan_name: str, utility: str | None = None, *, output_json: bool = False) → None

View a converted rate plan’s details.

async nwp500.cli.handle_tou_rates_request(zip_code: str, utility: str | None = None) → None

List utilities and rate plans for a zip code.

async nwp500.cli.handle_update_reservations_request(mqtt: NavienMqttClient, device: Device, reservations_json: str, enabled: bool) → None

Update reservation schedule.

nwp500.cli.load_tokens() → tuple[AuthTokens | None, str | None]

Load authentication tokens and user email from a file.

Returns:

Tuple of (AuthTokens, email) or (None, None) if tokens cannot be loaded

nwp500.cli.print_json(data: Any, indent: int = 2) → None

Print data as formatted JSON with optional syntax highlighting.

Uses Rich highlighting when available, falls back to plain JSON otherwise.

Parameters:

• data – Data to print

• indent – Number of spaces for indentation (default: 2)

nwp500.cli.save_tokens(tokens: AuthTokens, email: str) → None

Save authentication tokens and user email to a file.

Parameters:

• tokens – AuthTokens object containing credentials

• email – User email address

nwp500.cli.write_status_to_csv(file_path: str, status: DeviceStatus) → None

Append device status to a CSV file.

Parameters:

• file_path – Path to the CSV file

• status – DeviceStatus object to write