nwp500.cli package

Submodules

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

CSV and JSON renderers plus human-output dispatch for the CLI.

Data-shaping (which fields, labels, units, ordering, aggregation) lives in presentation; human-readable rendering is delegated to the Rich renderer in rich_output, which consumes the same neutral structures. This module additionally handles the genuinely different CSV and JSON outputs.

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 for a specific month via Rich.

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 (summary + monthly breakdown) via Rich.

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.presentation module

Presentation-neutral intermediate representation for CLI output.

This module owns the data-shaping concerns shared by every CLI output mode (Rich tables, CSV, JSON): which fields are shown, their labels, units, ordering and value-to-string formatting. Both the CSV/JSON renderer in output_formatters and the Rich renderer in rich_output consume the neutral structures produced here, so a field or command only needs to be described once.

The structures are deliberately free of any rendering technology (no colors, Rich objects or fixed-width layout). Renderers decide how to present them.

class nwp500.cli.presentation.DailyEnergyReport(year: int, month: int, totals: EnergyTotals, days: list[EnergyPeriodRow] = <factory>)

Bases: object

Aggregated daily energy usage for a single month.

days: list[EnergyPeriodRow]

month: int

totals: EnergyTotals

year: int

class nwp500.cli.presentation.EnergyPeriodRow(label: str, total_wh: int, heat_pump_wh: int, heat_element_wh: int, heat_pump_time: int, heat_element_time: int, heat_pump_percentage: float, heat_element_percentage: float)

Bases: object

A single aggregated energy period (a month or a day).

label identifies the period (“June 2025” for a month, "1" for a day). All usage values are in watt-hours; percentages are of total usage.

heat_element_percentage: float

heat_element_time: int

heat_element_wh: int

heat_pump_percentage: float

heat_pump_time: int

heat_pump_wh: int

label: str

total_wh: int

class nwp500.cli.presentation.EnergyReport(totals: EnergyTotals, months: list[EnergyPeriodRow] = <factory>)

Bases: object

Aggregated energy usage grouped by month.

months: list[EnergyPeriodRow]

totals: EnergyTotals

class nwp500.cli.presentation.EnergyTotals(total_usage_wh: int, heat_pump_usage_wh: int, heat_pump_percentage: float, heat_element_usage_wh: int, heat_element_percentage: float, total_time_hours: int, heat_pump_time_hours: int, heat_element_time_hours: int)

Bases: object

Presentation-neutral aggregated energy totals.

heat_element_percentage: float

heat_element_time_hours: int

heat_element_usage_wh: int

heat_pump_percentage: float

heat_pump_time_hours: int

heat_pump_usage_wh: int

total_time_hours: int

total_usage_wh: int

nwp500.cli.presentation.build_daily_energy_report(energy_response: Any, year: int, month: int) → DailyEnergyReport | None

Aggregate a single month’s daily energy into neutral rows.

Parameters:

• energy_response – EnergyUsageResponse object

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

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

Returns:

DailyEnergyReport, or None if the month has no data.

nwp500.cli.presentation.build_device_info_rows(device_feature: Any) → list[tuple[str, str, str]]

Build presentation-neutral rows for a device feature object.

Parameters:

device_feature – DeviceFeature object

Returns:

List of (section, label, value) rows in display order.

nwp500.cli.presentation.build_device_status_rows(device_status: Any) → list[tuple[str, str, str]]

Build presentation-neutral rows for a device status object.

Units are automatically extracted from the DeviceStatus model metadata.

Parameters:

device_status – DeviceStatus object

Returns:

List of (section, label, value) rows in display order.

nwp500.cli.presentation.build_energy_report(energy_response: Any) → EnergyReport

Aggregate an energy response into neutral monthly rows.

Parameters:

energy_response – EnergyUsageResponse object

Returns:

EnergyReport with totals and one row per month.

nwp500.cli.presentation.format_month_label(year: int, month: int) → str

Return a display label for a year/month pair.

nwp500.cli.rich_output module

Rich-based renderers for CLI human-readable output.

Rich is a hard requirement of the CLI; there is no plain-text fallback. This module renders the presentation-neutral structures built in presentation (and JSON/tables) using Rich exclusively.

class nwp500.cli.rich_output.OutputFormatter

Bases: object

Rich output formatter for CLI human-readable output.

Rich is mandatory; this formatter always renders with Rich.

print_daily_energy_table(report: DailyEnergyReport) → None

Print a daily energy usage report (summary + daily breakdown).

Parameters:

report – Neutral daily energy report from presentation

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(report: EnergyReport) → None

Print an energy usage report (summary + monthly breakdown).

Parameters:

report – Neutral energy report from presentation

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