diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 93b0ae01..8e6e4aab 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -5,6 +5,27 @@ Changelog Unreleased ========== +Changed +------- +- **CLI formatting stacks merged; Rich is the sole human renderer** + (`#100 `_): a new + ``src/nwp500/cli/presentation.py`` owns all data-shaping for CLI output + (field selection, labels, units, ordering, value formatting and energy + aggregation) as presentation-neutral structures. Because the CLI + hard-requires ``rich`` (there is no plain-text fallback), the redundant + plain-text human renderer and the Rich-vs-plain fallback machinery were + removed: ``rich_output.py`` no longer contains ``_should_use_rich``, + ``_rich_available``, the ``NWP500_NO_RICH`` toggle, or any + ``_print_*_plain`` methods, and now renders the neutral structures + (including the energy ``TOTAL SUMMARY``) exclusively with Rich, consuming + the presentation dataclasses directly instead of dict adapters. + ``output_formatters.py`` keeps only JSON/CSV rendering plus thin + human-output dispatch. Net CLI code drops from ~2088 to ~1838 lines with a + single data-shaping layer and a single human renderer. Non-energy output + is byte-for-byte unchanged (verified by golden capture and the existing + CLI tests); energy output now renders a summary table plus a breakdown + table entirely in Rich rather than printing plain text and a Rich table. + Fixed ----- - **MQTT ack futures now consumed on abandonment** (`#97 diff --git a/src/nwp500/cli/output_formatters.py b/src/nwp500/cli/output_formatters.py index bb5fde53..b960262f 100644 --- a/src/nwp500/cli/output_formatters.py +++ b/src/nwp500/cli/output_formatters.py @@ -1,94 +1,32 @@ -"""Output formatting utilities for CLI (CSV, JSON).""" +"""CSV and JSON renderers plus human-output dispatch for the CLI. + +Data-shaping (which fields, labels, units, ordering, aggregation) lives in +:mod:`.presentation`; human-readable rendering is delegated to the Rich +renderer in :mod:`.rich_output`, which consumes the same neutral structures. +This module additionally handles the genuinely different CSV and JSON outputs. +""" import csv import json import logging -from calendar import month_name from datetime import datetime from enum import Enum from pathlib import Path from typing import Any -from nwp500 import DeviceFeature, DeviceStatus +from nwp500 import DeviceStatus +from .presentation import ( + build_daily_energy_report, + build_device_info_rows, + build_device_status_rows, + build_energy_report, +) from .rich_output import get_formatter _logger = logging.getLogger(__name__) -def _format_number(value: Any) -> str: - """Format number to one decimal place if float, otherwise return as-is.""" - if isinstance(value, float): - return f"{value:.1f}" - return str(value) - - -def _get_unit_suffix( - field_name: str, - model_class: Any = DeviceStatus, - instance: Any = None, -) -> str: - """Extract unit suffix from model field metadata. - - For dynamic fields (temperature, flow_rate, water), use the instance's - get_field_unit() method to get the correct unit based on device preferences. - - Args: - field_name: Name of the field to get unit for - model_class: The Pydantic model class (default: DeviceStatus) - instance: Optional instance of the model for dynamic unit resolution - - Returns: - Unit string (e.g., "°F", "°C", "GPM", "Wh") or empty string if not found - """ - # Use instance's method if available for dynamic unit resolution - if instance and hasattr(instance, "get_field_unit"): - return instance.get_field_unit(field_name) - - # Fallback to static unit from schema - if not hasattr(model_class, "model_fields"): - return "" - - model_fields = model_class.model_fields - if field_name not in model_fields: - return "" - - field_info = model_fields[field_name] - if not hasattr(field_info, "json_schema_extra"): - return "" - - extra = field_info.json_schema_extra - if isinstance(extra, dict) and "unit_of_measurement" in extra: - unit_val = extra["unit_of_measurement"] - unit: str = unit_val if unit_val is not None else "" - return f" {unit}" if unit else "" - - return "" - - -def _add_numeric_item( - items: list[tuple[str, str, str]], - device_status: Any, - field_name: str, - category: str, - label: str, -) -> None: - """Add a numeric field with unit to items list, extracting unit from model. - - Args: - items: List to append to - device_status: DeviceStatus object - field_name: Name of the field to display - category: Category section in the output - label: Display label for the field - """ - if hasattr(device_status, field_name): - value = getattr(device_status, field_name) - unit = _get_unit_suffix(field_name, instance=device_status) - formatted = f"{_format_number(value)}{unit}" - items.append((category, label, formatted)) - - def _json_default_serializer(obj: Any) -> Any: """Serialize objects not serializable by default json code. @@ -115,244 +53,34 @@ def _json_default_serializer(obj: Any) -> Any: raise TypeError(f"Type {type(obj)} not serializable") -def format_energy_usage(energy_response: Any) -> str: - """ - Format energy usage response as a human-readable table. - - Args: - energy_response: EnergyUsageResponse object - - Returns: - Formatted string with energy usage data in tabular form - """ - lines: list[str] = [] - - # Add header - lines.append("=" * 90) - lines.append("ENERGY USAGE REPORT") - lines.append("=" * 90) - - # Total summary - total = energy_response.total - total_usage_wh = total.total_usage - total_time_hours = total.total_time - - lines.append("") - lines.append("TOTAL SUMMARY") - lines.append("-" * 90) - lines.append( - f"Total Energy Used: {total_usage_wh:,} Wh ({total_usage_wh / 1000:.2f} kWh)" # noqa: E501 - ) - lines.append( - f" Heat Pump: {total.heat_pump_usage:,} Wh ({total.heat_pump_percentage:.1f}%)" # noqa: E501 - ) - lines.append( - f" Heat Element: {total.heat_element_usage:,} Wh ({total.heat_element_percentage:.1f}%)" # noqa: E501 - ) - lines.append(f"Total Time Running: {total_time_hours} hours") - lines.append(f" Heat Pump: {total.heat_pump_time} hours") - lines.append(f" Heat Element: {total.heat_element_time} hours") - - # Monthly data - if energy_response.usage: - lines.append("") - lines.append("MONTHLY BREAKDOWN") - lines.append("-" * 90) - lines.append( - f"{'Month':<20} {'Energy (Wh)':<18} {'HP (Wh)':<15} {'HE (Wh)':<15} {'HP Time (h)':<15}" # noqa: E501 - ) - lines.append("-" * 90) - - for month_data in energy_response.usage: - month_name_str = ( - f"{month_name[month_data.month]} {month_data.year}" - if 1 <= month_data.month <= 12 - else f"Month {month_data.month} {month_data.year}" - ) - total_wh = sum( - d.heat_pump_usage + d.heat_element_usage - for d in month_data.data - ) - hp_wh = sum(d.heat_pump_usage for d in month_data.data) - he_wh = sum(d.heat_element_usage for d in month_data.data) - hp_time = sum(d.heat_pump_time for d in month_data.data) - - lines.append( - f"{month_name_str:<20} {total_wh:>16,} {hp_wh:>13,} {he_wh:>13,} {hp_time:>13}" # noqa: E501 - ) - - lines.append("=" * 90) - return "\n".join(lines) - - def 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. + """Print energy usage data (summary + monthly breakdown) via Rich. Args: energy_response: EnergyUsageResponse object """ - # First, print the plain text summary (always works) - print(format_energy_usage(energy_response)) - - # Also prepare and print rich table if available - months_data: list[dict[str, Any]] = [] - - if energy_response.usage: - for month_data in energy_response.usage: - month_name_str = ( - f"{month_name[month_data.month]} {month_data.year}" - if 1 <= month_data.month <= 12 - else f"Month {month_data.month} {month_data.year}" - ) - total_wh = sum( - d.heat_pump_usage + d.heat_element_usage - for d in month_data.data - ) - hp_wh = sum(d.heat_pump_usage for d in month_data.data) - he_wh = sum(d.heat_element_usage for d in month_data.data) - hp_pct = (hp_wh / total_wh * 100) if total_wh > 0 else 0 - he_pct = (he_wh / total_wh * 100) if total_wh > 0 else 0 - - months_data.append( - { - "month_str": month_name_str, - "total_kwh": total_wh / 1000, - "hp_kwh": hp_wh / 1000, - "hp_pct": hp_pct, - "he_kwh": he_wh / 1000, - "he_pct": he_pct, - } - ) - - # Print rich energy table if available - formatter = get_formatter() - formatter.print_energy_table(months_data) - - -def 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. - - Args: - 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 - """ - lines: list[str] = [] - - # Add header - lines.append("=" * 100) - month_str = ( - f"{month_name[month]} {year}" - if 1 <= month <= 12 - else f"Month {month} {year}" - ) - lines.append(f"DAILY ENERGY USAGE - {month_str}") - lines.append("=" * 100) - - # Find the month data - month_data = energy_response.get_month_data(year, month) - if not month_data or not month_data.data: - lines.append("No data available for this month") - lines.append("=" * 100) - return "\n".join(lines) - - # Total summary for the month - total = energy_response.total - total_usage_wh = total.total_usage - total_time_hours = total.total_time - - lines.append("") - lines.append("TOTAL SUMMARY") - lines.append("-" * 100) - lines.append( - f"Total Energy Used: {total_usage_wh:,} Wh ({total_usage_wh / 1000:.2f} kWh)" # noqa: E501 - ) - lines.append( - f" Heat Pump: {total.heat_pump_usage:,} Wh ({total.heat_pump_percentage:.1f}%)" # noqa: E501 - ) - lines.append( - f" Heat Element: {total.heat_element_usage:,} Wh ({total.heat_element_percentage:.1f}%)" # noqa: E501 - ) - lines.append(f"Total Time Running: {total_time_hours} hours") - lines.append(f" Heat Pump: {total.heat_pump_time} hours") - lines.append(f" Heat Element: {total.heat_element_time} hours") - - # Daily breakdown - lines.append("") - lines.append("DAILY BREAKDOWN") - lines.append("-" * 100) - lines.append( - f"{'Day':<5} {'Energy (Wh)':<18} {'HP (Wh)':<15} {'HE (Wh)':<15} {'HP Time':<12} {'HE Time':<12}" # noqa: E501 - ) - lines.append("-" * 100) - - for day_num, day_data in enumerate(month_data.data, start=1): - total_wh = day_data.total_usage - hp_wh = day_data.heat_pump_usage - he_wh = day_data.heat_element_usage - hp_time = day_data.heat_pump_time - he_time = day_data.heat_element_time - - lines.append( - f"{day_num:<5} {total_wh:>16,} {hp_wh:>13,} {he_wh:>13,} {hp_time:>10} {he_time:>10}" # noqa: E501 - ) - - lines.append("=" * 100) - return "\n".join(lines) + report = build_energy_report(energy_response) + get_formatter().print_energy_table(report) def 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. + """Print daily energy usage for a specific month via Rich. Args: energy_response: EnergyUsageResponse object year: Year to filter for (e.g., 2025) month: Month to filter for (1-12) """ - # First, print the plain text summary (always works) - print(format_daily_energy_usage(energy_response, year, month)) - - # Also prepare and print rich table if available - month_data = energy_response.get_month_data(year, month) - if not month_data or not month_data.data: - return - - days_data: list[dict[str, Any]] = [] - for day_num, day_data in enumerate(month_data.data, start=1): - total_wh = day_data.total_usage - hp_wh = day_data.heat_pump_usage - he_wh = day_data.heat_element_usage - hp_pct = (hp_wh / total_wh * 100) if total_wh > 0 else 0 - he_pct = (he_wh / total_wh * 100) if total_wh > 0 else 0 - - days_data.append( - { - "day": day_num, - "total_kwh": total_wh / 1000, - "hp_kwh": hp_wh / 1000, - "hp_pct": hp_pct, - "he_kwh": he_wh / 1000, - "he_pct": he_pct, - } - ) - - # Print rich energy table if available + report = build_daily_energy_report(energy_response, year, month) formatter = get_formatter() - formatter.print_daily_energy_table(days_data, year, month) + if report is None: + formatter.print_info( + f"No daily energy data available for {month}/{year}" + ) + return + formatter.print_daily_energy_table(report) def write_status_to_csv(file_path: str, status: DeviceStatus) -> None: @@ -429,476 +157,8 @@ def print_device_status(device_status: Any) -> None: Args: device_status: DeviceStatus object """ - # Collect all items with their categories - all_items: list[tuple[str, str, Any]] = [] - - # Operation Status - if hasattr(device_status, "operation_mode"): - mode = getattr( - device_status.operation_mode, "name", device_status.operation_mode - ) - all_items.append(("OPERATION STATUS", "Mode", mode)) - if hasattr(device_status, "operation_busy"): - all_items.append( - ( - "OPERATION STATUS", - "Busy", - "Yes" if device_status.operation_busy else "No", - ) - ) - if hasattr(device_status, "current_statenum"): - all_items.append( - ("OPERATION STATUS", "State", device_status.current_statenum) - ) - _add_numeric_item( - all_items, - device_status, - "current_inst_power", - "OPERATION STATUS", - "Current Power", - ) - - # Water Temperatures - _add_numeric_item( - all_items, - device_status, - "dhw_temperature", - "WATER TEMPERATURES", - "DHW Current", - ) - _add_numeric_item( - all_items, - device_status, - "dhw_target_temperature_setting", - "WATER TEMPERATURES", - "DHW Target", - ) - _add_numeric_item( - all_items, - device_status, - "tank_upper_temperature", - "WATER TEMPERATURES", - "Tank Upper", - ) - _add_numeric_item( - all_items, - device_status, - "tank_lower_temperature", - "WATER TEMPERATURES", - "Tank Lower", - ) - _add_numeric_item( - all_items, - device_status, - "current_inlet_temperature", - "WATER TEMPERATURES", - "Inlet Temp", - ) - _add_numeric_item( - all_items, - device_status, - "current_dhw_flow_rate", - "WATER TEMPERATURES", - "DHW Flow Rate", - ) - - # Ambient Temperatures - _add_numeric_item( - all_items, - device_status, - "outside_temperature", - "AMBIENT TEMPERATURES", - "Outside", - ) - _add_numeric_item( - all_items, - device_status, - "ambient_temperature", - "AMBIENT TEMPERATURES", - "Ambient", - ) - - # System Temperatures - _add_numeric_item( - all_items, - device_status, - "discharge_temperature", - "SYSTEM TEMPERATURES", - "Discharge", - ) - _add_numeric_item( - all_items, - device_status, - "suction_temperature", - "SYSTEM TEMPERATURES", - "Suction", - ) - _add_numeric_item( - all_items, - device_status, - "evaporator_temperature", - "SYSTEM TEMPERATURES", - "Evaporator", - ) - _add_numeric_item( - all_items, - device_status, - "target_super_heat", - "SYSTEM TEMPERATURES", - "Target SuperHeat", - ) - _add_numeric_item( - all_items, - device_status, - "current_super_heat", - "SYSTEM TEMPERATURES", - "Current SuperHeat", - ) - - # Heat Pump Settings - _add_numeric_item( - all_items, - device_status, - "hp_upper_on_temp_setting", - "HEAT PUMP SETTINGS", - "Upper On", - ) - _add_numeric_item( - all_items, - device_status, - "hp_upper_on_diff_temp_setting", - "HEAT PUMP SETTINGS", - "Upper On Diff", - ) - _add_numeric_item( - all_items, - device_status, - "hp_upper_off_temp_setting", - "HEAT PUMP SETTINGS", - "Upper Off", - ) - _add_numeric_item( - all_items, - device_status, - "hp_upper_off_diff_temp_setting", - "HEAT PUMP SETTINGS", - "Upper Off Diff", - ) - _add_numeric_item( - all_items, - device_status, - "hp_lower_on_temp_setting", - "HEAT PUMP SETTINGS", - "Lower On", - ) - _add_numeric_item( - all_items, - device_status, - "hp_lower_on_diff_temp_setting", - "HEAT PUMP SETTINGS", - "Lower On Diff", - ) - _add_numeric_item( - all_items, - device_status, - "hp_lower_off_temp_setting", - "HEAT PUMP SETTINGS", - "Lower Off", - ) - _add_numeric_item( - all_items, - device_status, - "hp_lower_off_diff_temp_setting", - "HEAT PUMP SETTINGS", - "Lower Off Diff", - ) - - # Heat Element Settings - _add_numeric_item( - all_items, - device_status, - "he_upper_on_temp_setting", - "HEAT ELEMENT SETTINGS", - "Upper On", - ) - _add_numeric_item( - all_items, - device_status, - "he_upper_on_diff_temp_setting", - "HEAT ELEMENT SETTINGS", - "Upper On Diff", - ) - _add_numeric_item( - all_items, - device_status, - "he_upper_off_temp_setting", - "HEAT ELEMENT SETTINGS", - "Upper Off", - ) - _add_numeric_item( - all_items, - device_status, - "he_upper_off_diff_temp_setting", - "HEAT ELEMENT SETTINGS", - "Upper Off Diff", - ) - _add_numeric_item( - all_items, - device_status, - "he_lower_on_temp_setting", - "HEAT ELEMENT SETTINGS", - "Lower On", - ) - _add_numeric_item( - all_items, - device_status, - "he_lower_on_diff_temp_setting", - "HEAT ELEMENT SETTINGS", - "Lower On Diff", - ) - _add_numeric_item( - all_items, - device_status, - "he_lower_off_temp_setting", - "HEAT ELEMENT SETTINGS", - "Lower Off", - ) - _add_numeric_item( - all_items, - device_status, - "he_lower_off_diff_temp_setting", - "HEAT ELEMENT SETTINGS", - "Lower Off Diff", - ) - - # Power & Energy - if hasattr(device_status, "wh_total_power_consumption"): - all_items.append( - ( - "POWER & ENERGY", - "Total Consumption", - f"{_format_number(device_status.wh_total_power_consumption)}Wh", - ) - ) - if hasattr(device_status, "wh_heat_pump_power"): - all_items.append( - ( - "POWER & ENERGY", - "Heat Pump Power", - f"{_format_number(device_status.wh_heat_pump_power)}Wh", - ) - ) - if hasattr(device_status, "wh_electric_heater_power"): - all_items.append( - ( - "POWER & ENERGY", - "Electric Heater Power", - f"{_format_number(device_status.wh_electric_heater_power)}Wh", - ) - ) - _add_numeric_item( - all_items, - device_status, - "total_energy_capacity", - "POWER & ENERGY", - "Total Capacity", - ) - _add_numeric_item( - all_items, - device_status, - "available_energy_capacity", - "POWER & ENERGY", - "Available Capacity", - ) - - # Fan Control - _add_numeric_item( - all_items, device_status, "target_fan_rpm", "FAN CONTROL", "Target RPM" - ) - _add_numeric_item( - all_items, - device_status, - "current_fan_rpm", - "FAN CONTROL", - "Current RPM", - ) - if hasattr(device_status, "fan_pwm"): - pwm_pct = f"{_format_number(device_status.fan_pwm)}%" - all_items.append(("FAN CONTROL", "PWM", pwm_pct)) - _add_numeric_item( - all_items, - device_status, - "cumulated_op_time_eva_fan", - "FAN CONTROL", - "Eva Fan Time", - ) - - # Compressor & Valve - if hasattr(device_status, "mixing_rate"): - mixing = f"{_format_number(device_status.mixing_rate)}%" - all_items.append(("COMPRESSOR & VALVE", "Mixing Rate", mixing)) - if hasattr(device_status, "eev_step"): - eev = f"{_format_number(device_status.eev_step)} steps" - all_items.append(("COMPRESSOR & VALVE", "EEV Step", eev)) - _add_numeric_item( - all_items, - device_status, - "target_super_heat", - "COMPRESSOR & VALVE", - "Target SuperHeat", - ) - _add_numeric_item( - all_items, - device_status, - "current_super_heat", - "COMPRESSOR & VALVE", - "Current SuperHeat", - ) - - # Recirculation - if hasattr(device_status, "recirc_operation_mode"): - all_items.append( - ( - "RECIRCULATION", - "Operation Mode", - device_status.recirc_operation_mode, - ) - ) - if hasattr(device_status, "recirc_pump_operation_status"): - all_items.append( - ( - "RECIRCULATION", - "Pump Status", - device_status.recirc_pump_operation_status, - ) - ) - _add_numeric_item( - all_items, - device_status, - "recirc_temperature", - "RECIRCULATION", - "Temperature", - ) - _add_numeric_item( - all_items, - device_status, - "recirc_faucet_temperature", - "RECIRCULATION", - "Faucet Temp", - ) - - # Status & Alerts - if hasattr(device_status, "error_code"): - all_items.append( - ("STATUS & ALERTS", "Error Code", device_status.error_code) - ) - if hasattr(device_status, "sub_error_code"): - all_items.append( - ("STATUS & ALERTS", "Sub Error Code", device_status.sub_error_code) - ) - if hasattr(device_status, "fault_status1"): - all_items.append( - ("STATUS & ALERTS", "Fault Status 1", device_status.fault_status1) - ) - if hasattr(device_status, "fault_status2"): - all_items.append( - ("STATUS & ALERTS", "Fault Status 2", device_status.fault_status2) - ) - if hasattr(device_status, "error_buzzer_use"): - all_items.append( - ( - "STATUS & ALERTS", - "Error Buzzer", - "Yes" if device_status.error_buzzer_use else "No", - ) - ) - - # Vacation Mode - _add_numeric_item( - all_items, - device_status, - "vacation_day_setting", - "VACATION MODE", - "Days Set", - ) - _add_numeric_item( - all_items, - device_status, - "vacation_day_elapsed", - "VACATION MODE", - "Days Elapsed", - ) - - # Air Filter - _add_numeric_item( - all_items, - device_status, - "air_filter_alarm_period", - "AIR FILTER", - "Alarm Period", - ) - _add_numeric_item( - all_items, - device_status, - "air_filter_alarm_elapsed", - "AIR FILTER", - "Alarm Elapsed", - ) - - # WiFi & Network - _add_numeric_item( - all_items, device_status, "wifi_rssi", "WiFi & NETWORK", "RSSI" - ) - - # Demand Response & TOU - if hasattr(device_status, "dr_event_status"): - all_items.append( - ( - "DEMAND RESPONSE & TOU", - "DR Event Status", - device_status.dr_event_status, - ) - ) - _add_numeric_item( - all_items, - device_status, - "dr_override_status", - "DEMAND RESPONSE & TOU", - "DR Override Status", - ) - if hasattr(device_status, "tou_status"): - all_items.append( - ("DEMAND RESPONSE & TOU", "TOU Status", device_status.tou_status) - ) - if hasattr(device_status, "tou_override_status"): - all_items.append( - ( - "DEMAND RESPONSE & TOU", - "TOU Override Status", - device_status.tou_override_status, - ) - ) - - # Anti-Legionella - _add_numeric_item( - all_items, - device_status, - "anti_legionella_period", - "ANTI-LEGIONELLA", - "Period", - ) - if hasattr(device_status, "anti_legionella_operation_busy"): - all_items.append( - ( - "ANTI-LEGIONELLA", - "Operation Busy", - "Yes" if device_status.anti_legionella_operation_busy else "No", - ) - ) - - # Use rich formatter for output formatter = get_formatter() - formatter.print_status_table(all_items) + formatter.print_status_table(build_device_status_rows(device_status)) def print_device_info(device_feature: Any) -> None: @@ -908,199 +168,5 @@ def print_device_info(device_feature: Any) -> None: Args: device_feature: DeviceFeature object """ - # Serialize to dict to get enum names from model_dump() - if hasattr(device_feature, "model_dump"): - device_dict = device_feature.model_dump() - else: - device_dict = device_feature - - # Collect all items with their categories - all_items: list[tuple[str, str, Any]] = [] - - # Device Identity - if "controller_serial_number" in device_dict: - all_items.append( - ( - "DEVICE IDENTITY", - "Serial Number", - device_dict["controller_serial_number"], - ) - ) - if "country_code" in device_dict: - all_items.append( - ("DEVICE IDENTITY", "Country Code", device_dict["country_code"]) - ) - if "model_type_code" in device_dict: - all_items.append( - ("DEVICE IDENTITY", "Model Type", device_dict["model_type_code"]) - ) - if "control_type_code" in device_dict: - all_items.append( - ( - "DEVICE IDENTITY", - "Control Type", - device_dict["control_type_code"], - ) - ) - if "volume_code" in device_dict: - all_items.append( - ("DEVICE IDENTITY", "Volume Code", device_dict["volume_code"]) - ) - - # Firmware Versions - if "controller_sw_version" in device_dict: - all_items.append( - ( - "FIRMWARE VERSIONS", - "Controller Version", - f"v{device_dict['controller_sw_version']}", - ) - ) - if "controller_sw_code" in device_dict: - all_items.append( - ( - "FIRMWARE VERSIONS", - "Controller Code", - device_dict["controller_sw_code"], - ) - ) - if "panel_sw_version" in device_dict: - all_items.append( - ( - "FIRMWARE VERSIONS", - "Panel Version", - f"v{device_dict['panel_sw_version']}", - ) - ) - if "panel_sw_code" in device_dict: - all_items.append( - ("FIRMWARE VERSIONS", "Panel Code", device_dict["panel_sw_code"]) - ) - if "wifi_sw_version" in device_dict: - all_items.append( - ( - "FIRMWARE VERSIONS", - "WiFi Version", - f"v{device_dict['wifi_sw_version']}", - ) - ) - if "wifi_sw_code" in device_dict: - all_items.append( - ("FIRMWARE VERSIONS", "WiFi Code", device_dict["wifi_sw_code"]) - ) - if ( - hasattr(device_feature, "recirc_sw_version") - and device_dict["recirc_sw_version"] > 0 - ): - all_items.append( - ( - "FIRMWARE VERSIONS", - "Recirculation Version", - f"v{device_dict['recirc_sw_version']}", - ) - ) - if "recirc_model_type_code" in device_dict: - all_items.append( - ( - "FIRMWARE VERSIONS", - "Recirculation Model", - device_dict["recirc_model_type_code"], - ) - ) - - # Configuration - if "temperature_type" in device_dict: - temp_type = getattr( - device_dict["temperature_type"], - "name", - device_dict["temperature_type"], - ) - all_items.append(("CONFIGURATION", "Temperature Unit", temp_type)) - if "temp_formula_type" in device_dict: - all_items.append( - ( - "CONFIGURATION", - "Temperature Formula", - device_dict["temp_formula_type"], - ) - ) - if "dhw_temperature_min" in device_dict: - unit_suffix = ( - _get_unit_suffix( - "dhw_temperature_min", DeviceFeature, device_feature - ) - if hasattr(device_feature, "get_field_unit") - else " °F" - ) - all_items.append( - ( - "CONFIGURATION", - "DHW Temp Range", - f"{device_dict['dhw_temperature_min']}{unit_suffix} - {device_dict['dhw_temperature_max']}{unit_suffix}", # noqa: E501 - ) - ) - if "freeze_protection_temp_min" in device_dict: - unit_suffix = ( - _get_unit_suffix( - "freeze_protection_temp_min", DeviceFeature, device_feature - ) - if hasattr(device_feature, "get_field_unit") - else " °F" - ) - all_items.append( - ( - "CONFIGURATION", - "Freeze Protection Range", - f"{device_dict['freeze_protection_temp_min']}{unit_suffix} - {device_dict['freeze_protection_temp_max']}{unit_suffix}", # noqa: E501 - ) - ) - if "recirc_temperature_min" in device_dict: - unit_suffix = ( - _get_unit_suffix( - "recirc_temperature_min", DeviceFeature, device_feature - ) - if hasattr(device_feature, "get_field_unit") - else " °F" - ) - all_items.append( - ( - "CONFIGURATION", - "Recirculation Temp Range", - f"{device_dict['recirc_temperature_min']}{unit_suffix} - {device_dict['recirc_temperature_max']}{unit_suffix}", # noqa: E501 - ) - ) - - # Supported Features - features_list = [ - ("Power Control", "power_use"), - ("DHW Control", "dhw_use"), - ("Heat Pump Mode", "heatpump_use"), - ("Electric Mode", "electric_use"), - ("Energy Saver", "energy_saver_use"), - ("High Demand", "high_demand_use"), - ("Eco Mode", "eco_use"), - ("Holiday Mode", "holiday_use"), - ("Program Reservation", "program_reservation_use"), - ("Recirculation", "recirculation_use"), - ("Recirculation Reservation", "recirc_reservation_use"), - ("Smart Diagnostic", "smart_diagnostic_use"), - ("WiFi RSSI", "wifi_rssi_use"), - ("Energy Usage", "energy_usage_use"), - ("Freeze Protection", "freeze_protection_use"), - ("Mixing Valve", "mixing_valve_use"), - ("DR Settings", "dr_setting_use"), - ("Anti-Legionella", "anti_legionella_setting_use"), - ("HPWH", "hpwh_use"), - ("DHW Refill", "dhw_refill_use"), - ("Title 24", "title24_use"), - ] - - for label, attr in features_list: - if hasattr(device_feature, attr): - value = getattr(device_feature, attr) - status = "Yes" if value else "No" - all_items.append(("SUPPORTED FEATURES", label, status)) - - # Use rich formatter for output formatter = get_formatter() - formatter.print_status_table(all_items) + formatter.print_status_table(build_device_info_rows(device_feature)) diff --git a/src/nwp500/cli/presentation.py b/src/nwp500/cli/presentation.py new file mode 100644 index 00000000..8fcbf9dc --- /dev/null +++ b/src/nwp500/cli/presentation.py @@ -0,0 +1,937 @@ +"""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 +:mod:`.output_formatters` and the Rich renderer in :mod:`.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. +""" + +from calendar import month_name +from dataclasses import dataclass, field +from typing import Any + +from nwp500 import DeviceFeature, DeviceStatus + +# A single labeled row within a titled section: +# ``(section, label, value)`` where ``value`` is already formatted to a string. +StatusRow = tuple[str, str, str] + + +def _format_number(value: Any) -> str: + """Format number to one decimal place if float, otherwise return as-is.""" + if isinstance(value, float): + return f"{value:.1f}" + return str(value) + + +def _get_unit_suffix( + field_name: str, + model_class: Any = DeviceStatus, + instance: Any = None, +) -> str: + """Extract unit suffix from model field metadata. + + For dynamic fields (temperature, flow_rate, water), use the instance's + get_field_unit() method to get the correct unit based on device preferences. + + Args: + field_name: Name of the field to get unit for + model_class: The Pydantic model class (default: DeviceStatus) + instance: Optional instance of the model for dynamic unit resolution + + Returns: + Unit string (e.g., "°F", "°C", "GPM", "Wh") or empty string if not found + """ + # Use instance's method if available for dynamic unit resolution + if instance and hasattr(instance, "get_field_unit"): + return instance.get_field_unit(field_name) + + # Fallback to static unit from schema + if not hasattr(model_class, "model_fields"): + return "" + + model_fields = model_class.model_fields + if field_name not in model_fields: + return "" + + field_info = model_fields[field_name] + if not hasattr(field_info, "json_schema_extra"): + return "" + + extra = field_info.json_schema_extra + if isinstance(extra, dict) and "unit_of_measurement" in extra: + unit_val = extra["unit_of_measurement"] + unit: str = unit_val if unit_val is not None else "" + return f" {unit}" if unit else "" + + return "" + + +def _add_numeric_item( + items: list[StatusRow], + device_status: Any, + field_name: str, + category: str, + label: str, +) -> None: + """Add a numeric field with unit to items list, extracting unit from model. + + Args: + items: List to append to + device_status: DeviceStatus object + field_name: Name of the field to display + category: Category section in the output + label: Display label for the field + """ + if hasattr(device_status, field_name): + value = getattr(device_status, field_name) + unit = _get_unit_suffix(field_name, instance=device_status) + formatted = f"{_format_number(value)}{unit}" + items.append((category, label, formatted)) + + +def build_device_status_rows(device_status: Any) -> list[StatusRow]: + """Build presentation-neutral rows for a device status object. + + Units are automatically extracted from the DeviceStatus model metadata. + + Args: + device_status: DeviceStatus object + + Returns: + List of ``(section, label, value)`` rows in display order. + """ + all_items: list[StatusRow] = [] + + # Operation Status + if hasattr(device_status, "operation_mode"): + mode = getattr( + device_status.operation_mode, "name", device_status.operation_mode + ) + all_items.append(("OPERATION STATUS", "Mode", mode)) + if hasattr(device_status, "operation_busy"): + all_items.append( + ( + "OPERATION STATUS", + "Busy", + "Yes" if device_status.operation_busy else "No", + ) + ) + if hasattr(device_status, "current_statenum"): + all_items.append( + ("OPERATION STATUS", "State", device_status.current_statenum) + ) + _add_numeric_item( + all_items, + device_status, + "current_inst_power", + "OPERATION STATUS", + "Current Power", + ) + + # Water Temperatures + _add_numeric_item( + all_items, + device_status, + "dhw_temperature", + "WATER TEMPERATURES", + "DHW Current", + ) + _add_numeric_item( + all_items, + device_status, + "dhw_target_temperature_setting", + "WATER TEMPERATURES", + "DHW Target", + ) + _add_numeric_item( + all_items, + device_status, + "tank_upper_temperature", + "WATER TEMPERATURES", + "Tank Upper", + ) + _add_numeric_item( + all_items, + device_status, + "tank_lower_temperature", + "WATER TEMPERATURES", + "Tank Lower", + ) + _add_numeric_item( + all_items, + device_status, + "current_inlet_temperature", + "WATER TEMPERATURES", + "Inlet Temp", + ) + _add_numeric_item( + all_items, + device_status, + "current_dhw_flow_rate", + "WATER TEMPERATURES", + "DHW Flow Rate", + ) + + # Ambient Temperatures + _add_numeric_item( + all_items, + device_status, + "outside_temperature", + "AMBIENT TEMPERATURES", + "Outside", + ) + _add_numeric_item( + all_items, + device_status, + "ambient_temperature", + "AMBIENT TEMPERATURES", + "Ambient", + ) + + # System Temperatures + _add_numeric_item( + all_items, + device_status, + "discharge_temperature", + "SYSTEM TEMPERATURES", + "Discharge", + ) + _add_numeric_item( + all_items, + device_status, + "suction_temperature", + "SYSTEM TEMPERATURES", + "Suction", + ) + _add_numeric_item( + all_items, + device_status, + "evaporator_temperature", + "SYSTEM TEMPERATURES", + "Evaporator", + ) + _add_numeric_item( + all_items, + device_status, + "target_super_heat", + "SYSTEM TEMPERATURES", + "Target SuperHeat", + ) + _add_numeric_item( + all_items, + device_status, + "current_super_heat", + "SYSTEM TEMPERATURES", + "Current SuperHeat", + ) + + # Heat Pump Settings + _add_numeric_item( + all_items, + device_status, + "hp_upper_on_temp_setting", + "HEAT PUMP SETTINGS", + "Upper On", + ) + _add_numeric_item( + all_items, + device_status, + "hp_upper_on_diff_temp_setting", + "HEAT PUMP SETTINGS", + "Upper On Diff", + ) + _add_numeric_item( + all_items, + device_status, + "hp_upper_off_temp_setting", + "HEAT PUMP SETTINGS", + "Upper Off", + ) + _add_numeric_item( + all_items, + device_status, + "hp_upper_off_diff_temp_setting", + "HEAT PUMP SETTINGS", + "Upper Off Diff", + ) + _add_numeric_item( + all_items, + device_status, + "hp_lower_on_temp_setting", + "HEAT PUMP SETTINGS", + "Lower On", + ) + _add_numeric_item( + all_items, + device_status, + "hp_lower_on_diff_temp_setting", + "HEAT PUMP SETTINGS", + "Lower On Diff", + ) + _add_numeric_item( + all_items, + device_status, + "hp_lower_off_temp_setting", + "HEAT PUMP SETTINGS", + "Lower Off", + ) + _add_numeric_item( + all_items, + device_status, + "hp_lower_off_diff_temp_setting", + "HEAT PUMP SETTINGS", + "Lower Off Diff", + ) + + # Heat Element Settings + _add_numeric_item( + all_items, + device_status, + "he_upper_on_temp_setting", + "HEAT ELEMENT SETTINGS", + "Upper On", + ) + _add_numeric_item( + all_items, + device_status, + "he_upper_on_diff_temp_setting", + "HEAT ELEMENT SETTINGS", + "Upper On Diff", + ) + _add_numeric_item( + all_items, + device_status, + "he_upper_off_temp_setting", + "HEAT ELEMENT SETTINGS", + "Upper Off", + ) + _add_numeric_item( + all_items, + device_status, + "he_upper_off_diff_temp_setting", + "HEAT ELEMENT SETTINGS", + "Upper Off Diff", + ) + _add_numeric_item( + all_items, + device_status, + "he_lower_on_temp_setting", + "HEAT ELEMENT SETTINGS", + "Lower On", + ) + _add_numeric_item( + all_items, + device_status, + "he_lower_on_diff_temp_setting", + "HEAT ELEMENT SETTINGS", + "Lower On Diff", + ) + _add_numeric_item( + all_items, + device_status, + "he_lower_off_temp_setting", + "HEAT ELEMENT SETTINGS", + "Lower Off", + ) + _add_numeric_item( + all_items, + device_status, + "he_lower_off_diff_temp_setting", + "HEAT ELEMENT SETTINGS", + "Lower Off Diff", + ) + + # Power & Energy + if hasattr(device_status, "wh_total_power_consumption"): + all_items.append( + ( + "POWER & ENERGY", + "Total Consumption", + f"{_format_number(device_status.wh_total_power_consumption)}Wh", + ) + ) + if hasattr(device_status, "wh_heat_pump_power"): + all_items.append( + ( + "POWER & ENERGY", + "Heat Pump Power", + f"{_format_number(device_status.wh_heat_pump_power)}Wh", + ) + ) + if hasattr(device_status, "wh_electric_heater_power"): + all_items.append( + ( + "POWER & ENERGY", + "Electric Heater Power", + f"{_format_number(device_status.wh_electric_heater_power)}Wh", + ) + ) + _add_numeric_item( + all_items, + device_status, + "total_energy_capacity", + "POWER & ENERGY", + "Total Capacity", + ) + _add_numeric_item( + all_items, + device_status, + "available_energy_capacity", + "POWER & ENERGY", + "Available Capacity", + ) + + # Fan Control + _add_numeric_item( + all_items, device_status, "target_fan_rpm", "FAN CONTROL", "Target RPM" + ) + _add_numeric_item( + all_items, + device_status, + "current_fan_rpm", + "FAN CONTROL", + "Current RPM", + ) + if hasattr(device_status, "fan_pwm"): + pwm_pct = f"{_format_number(device_status.fan_pwm)}%" + all_items.append(("FAN CONTROL", "PWM", pwm_pct)) + _add_numeric_item( + all_items, + device_status, + "cumulated_op_time_eva_fan", + "FAN CONTROL", + "Eva Fan Time", + ) + + # Compressor & Valve + if hasattr(device_status, "mixing_rate"): + mixing = f"{_format_number(device_status.mixing_rate)}%" + all_items.append(("COMPRESSOR & VALVE", "Mixing Rate", mixing)) + if hasattr(device_status, "eev_step"): + eev = f"{_format_number(device_status.eev_step)} steps" + all_items.append(("COMPRESSOR & VALVE", "EEV Step", eev)) + _add_numeric_item( + all_items, + device_status, + "target_super_heat", + "COMPRESSOR & VALVE", + "Target SuperHeat", + ) + _add_numeric_item( + all_items, + device_status, + "current_super_heat", + "COMPRESSOR & VALVE", + "Current SuperHeat", + ) + + # Recirculation + if hasattr(device_status, "recirc_operation_mode"): + all_items.append( + ( + "RECIRCULATION", + "Operation Mode", + device_status.recirc_operation_mode, + ) + ) + if hasattr(device_status, "recirc_pump_operation_status"): + all_items.append( + ( + "RECIRCULATION", + "Pump Status", + device_status.recirc_pump_operation_status, + ) + ) + _add_numeric_item( + all_items, + device_status, + "recirc_temperature", + "RECIRCULATION", + "Temperature", + ) + _add_numeric_item( + all_items, + device_status, + "recirc_faucet_temperature", + "RECIRCULATION", + "Faucet Temp", + ) + + # Status & Alerts + if hasattr(device_status, "error_code"): + all_items.append( + ("STATUS & ALERTS", "Error Code", device_status.error_code) + ) + if hasattr(device_status, "sub_error_code"): + all_items.append( + ("STATUS & ALERTS", "Sub Error Code", device_status.sub_error_code) + ) + if hasattr(device_status, "fault_status1"): + all_items.append( + ("STATUS & ALERTS", "Fault Status 1", device_status.fault_status1) + ) + if hasattr(device_status, "fault_status2"): + all_items.append( + ("STATUS & ALERTS", "Fault Status 2", device_status.fault_status2) + ) + if hasattr(device_status, "error_buzzer_use"): + all_items.append( + ( + "STATUS & ALERTS", + "Error Buzzer", + "Yes" if device_status.error_buzzer_use else "No", + ) + ) + + # Vacation Mode + _add_numeric_item( + all_items, + device_status, + "vacation_day_setting", + "VACATION MODE", + "Days Set", + ) + _add_numeric_item( + all_items, + device_status, + "vacation_day_elapsed", + "VACATION MODE", + "Days Elapsed", + ) + + # Air Filter + _add_numeric_item( + all_items, + device_status, + "air_filter_alarm_period", + "AIR FILTER", + "Alarm Period", + ) + _add_numeric_item( + all_items, + device_status, + "air_filter_alarm_elapsed", + "AIR FILTER", + "Alarm Elapsed", + ) + + # WiFi & Network + _add_numeric_item( + all_items, device_status, "wifi_rssi", "WiFi & NETWORK", "RSSI" + ) + + # Demand Response & TOU + if hasattr(device_status, "dr_event_status"): + all_items.append( + ( + "DEMAND RESPONSE & TOU", + "DR Event Status", + device_status.dr_event_status, + ) + ) + _add_numeric_item( + all_items, + device_status, + "dr_override_status", + "DEMAND RESPONSE & TOU", + "DR Override Status", + ) + if hasattr(device_status, "tou_status"): + all_items.append( + ("DEMAND RESPONSE & TOU", "TOU Status", device_status.tou_status) + ) + if hasattr(device_status, "tou_override_status"): + all_items.append( + ( + "DEMAND RESPONSE & TOU", + "TOU Override Status", + device_status.tou_override_status, + ) + ) + + # Anti-Legionella + _add_numeric_item( + all_items, + device_status, + "anti_legionella_period", + "ANTI-LEGIONELLA", + "Period", + ) + if hasattr(device_status, "anti_legionella_operation_busy"): + all_items.append( + ( + "ANTI-LEGIONELLA", + "Operation Busy", + "Yes" if device_status.anti_legionella_operation_busy else "No", + ) + ) + + return all_items + + +def build_device_info_rows(device_feature: Any) -> list[StatusRow]: + """Build presentation-neutral rows for a device feature object. + + Args: + device_feature: DeviceFeature object + + Returns: + List of ``(section, label, value)`` rows in display order. + """ + # Serialize to dict to get enum names from model_dump() + if hasattr(device_feature, "model_dump"): + device_dict = device_feature.model_dump() + else: + device_dict = device_feature + + all_items: list[StatusRow] = [] + + # Device Identity + if "controller_serial_number" in device_dict: + all_items.append( + ( + "DEVICE IDENTITY", + "Serial Number", + device_dict["controller_serial_number"], + ) + ) + if "country_code" in device_dict: + all_items.append( + ("DEVICE IDENTITY", "Country Code", device_dict["country_code"]) + ) + if "model_type_code" in device_dict: + all_items.append( + ("DEVICE IDENTITY", "Model Type", device_dict["model_type_code"]) + ) + if "control_type_code" in device_dict: + all_items.append( + ( + "DEVICE IDENTITY", + "Control Type", + device_dict["control_type_code"], + ) + ) + if "volume_code" in device_dict: + all_items.append( + ("DEVICE IDENTITY", "Volume Code", device_dict["volume_code"]) + ) + + # Firmware Versions + if "controller_sw_version" in device_dict: + all_items.append( + ( + "FIRMWARE VERSIONS", + "Controller Version", + f"v{device_dict['controller_sw_version']}", + ) + ) + if "controller_sw_code" in device_dict: + all_items.append( + ( + "FIRMWARE VERSIONS", + "Controller Code", + device_dict["controller_sw_code"], + ) + ) + if "panel_sw_version" in device_dict: + all_items.append( + ( + "FIRMWARE VERSIONS", + "Panel Version", + f"v{device_dict['panel_sw_version']}", + ) + ) + if "panel_sw_code" in device_dict: + all_items.append( + ("FIRMWARE VERSIONS", "Panel Code", device_dict["panel_sw_code"]) + ) + if "wifi_sw_version" in device_dict: + all_items.append( + ( + "FIRMWARE VERSIONS", + "WiFi Version", + f"v{device_dict['wifi_sw_version']}", + ) + ) + if "wifi_sw_code" in device_dict: + all_items.append( + ("FIRMWARE VERSIONS", "WiFi Code", device_dict["wifi_sw_code"]) + ) + if ( + hasattr(device_feature, "recirc_sw_version") + and device_dict["recirc_sw_version"] > 0 + ): + all_items.append( + ( + "FIRMWARE VERSIONS", + "Recirculation Version", + f"v{device_dict['recirc_sw_version']}", + ) + ) + if "recirc_model_type_code" in device_dict: + all_items.append( + ( + "FIRMWARE VERSIONS", + "Recirculation Model", + device_dict["recirc_model_type_code"], + ) + ) + + # Configuration + if "temperature_type" in device_dict: + temp_type = getattr( + device_dict["temperature_type"], + "name", + device_dict["temperature_type"], + ) + all_items.append(("CONFIGURATION", "Temperature Unit", temp_type)) + if "temp_formula_type" in device_dict: + all_items.append( + ( + "CONFIGURATION", + "Temperature Formula", + device_dict["temp_formula_type"], + ) + ) + if "dhw_temperature_min" in device_dict: + unit_suffix = ( + _get_unit_suffix( + "dhw_temperature_min", DeviceFeature, device_feature + ) + if hasattr(device_feature, "get_field_unit") + else " °F" + ) + all_items.append( + ( + "CONFIGURATION", + "DHW Temp Range", + f"{device_dict['dhw_temperature_min']}{unit_suffix} - {device_dict['dhw_temperature_max']}{unit_suffix}", # noqa: E501 + ) + ) + if "freeze_protection_temp_min" in device_dict: + unit_suffix = ( + _get_unit_suffix( + "freeze_protection_temp_min", DeviceFeature, device_feature + ) + if hasattr(device_feature, "get_field_unit") + else " °F" + ) + all_items.append( + ( + "CONFIGURATION", + "Freeze Protection Range", + f"{device_dict['freeze_protection_temp_min']}{unit_suffix} - {device_dict['freeze_protection_temp_max']}{unit_suffix}", # noqa: E501 + ) + ) + if "recirc_temperature_min" in device_dict: + unit_suffix = ( + _get_unit_suffix( + "recirc_temperature_min", DeviceFeature, device_feature + ) + if hasattr(device_feature, "get_field_unit") + else " °F" + ) + all_items.append( + ( + "CONFIGURATION", + "Recirculation Temp Range", + f"{device_dict['recirc_temperature_min']}{unit_suffix} - {device_dict['recirc_temperature_max']}{unit_suffix}", # noqa: E501 + ) + ) + + # Supported Features + features_list = [ + ("Power Control", "power_use"), + ("DHW Control", "dhw_use"), + ("Heat Pump Mode", "heatpump_use"), + ("Electric Mode", "electric_use"), + ("Energy Saver", "energy_saver_use"), + ("High Demand", "high_demand_use"), + ("Eco Mode", "eco_use"), + ("Holiday Mode", "holiday_use"), + ("Program Reservation", "program_reservation_use"), + ("Recirculation", "recirculation_use"), + ("Recirculation Reservation", "recirc_reservation_use"), + ("Smart Diagnostic", "smart_diagnostic_use"), + ("WiFi RSSI", "wifi_rssi_use"), + ("Energy Usage", "energy_usage_use"), + ("Freeze Protection", "freeze_protection_use"), + ("Mixing Valve", "mixing_valve_use"), + ("DR Settings", "dr_setting_use"), + ("Anti-Legionella", "anti_legionella_setting_use"), + ("HPWH", "hpwh_use"), + ("DHW Refill", "dhw_refill_use"), + ("Title 24", "title24_use"), + ] + + for label, attr in features_list: + if hasattr(device_feature, attr): + value = getattr(device_feature, attr) + status = "Yes" if value else "No" + all_items.append(("SUPPORTED FEATURES", label, status)) + + return all_items + + +def format_month_label(year: int, month: int) -> str: + """Return a display label for a year/month pair.""" + if 1 <= month <= 12: + return f"{month_name[month]} {year}" + return f"Month {month} {year}" + + +@dataclass +class EnergyTotals: + """Presentation-neutral aggregated energy totals.""" + + 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 + + +@dataclass +class EnergyPeriodRow: + """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. + """ + + 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 + + +@dataclass +class EnergyReport: + """Aggregated energy usage grouped by month.""" + + totals: EnergyTotals + months: list[EnergyPeriodRow] = field(default_factory=list) + + +@dataclass +class DailyEnergyReport: + """Aggregated daily energy usage for a single month.""" + + year: int + month: int + totals: EnergyTotals + days: list[EnergyPeriodRow] = field(default_factory=list) + + +def _build_totals(total: Any) -> EnergyTotals: + """Build neutral totals from an EnergyUsageTotal model.""" + return EnergyTotals( + total_usage_wh=total.total_usage, + heat_pump_usage_wh=total.heat_pump_usage, + heat_pump_percentage=total.heat_pump_percentage, + heat_element_usage_wh=total.heat_element_usage, + heat_element_percentage=total.heat_element_percentage, + total_time_hours=total.total_time, + heat_pump_time_hours=total.heat_pump_time, + heat_element_time_hours=total.heat_element_time, + ) + + +def _percentages(hp_wh: int, he_wh: int, total_wh: int) -> tuple[float, float]: + """Return heat-pump/heat-element share of total usage as percentages.""" + hp_pct = (hp_wh / total_wh * 100) if total_wh > 0 else 0.0 + he_pct = (he_wh / total_wh * 100) if total_wh > 0 else 0.0 + return hp_pct, he_pct + + +def build_energy_report(energy_response: Any) -> EnergyReport: + """Aggregate an energy response into neutral monthly rows. + + Args: + energy_response: EnergyUsageResponse object + + Returns: + EnergyReport with totals and one row per month. + """ + months: list[EnergyPeriodRow] = [] + for month_data in energy_response.usage: + label = format_month_label(month_data.year, month_data.month) + total_wh = sum( + d.heat_pump_usage + d.heat_element_usage for d in month_data.data + ) + hp_wh = sum(d.heat_pump_usage for d in month_data.data) + he_wh = sum(d.heat_element_usage for d in month_data.data) + hp_time = sum(d.heat_pump_time for d in month_data.data) + he_time = sum(d.heat_element_time for d in month_data.data) + hp_pct, he_pct = _percentages(hp_wh, he_wh, total_wh) + months.append( + EnergyPeriodRow( + label=label, + total_wh=total_wh, + heat_pump_wh=hp_wh, + heat_element_wh=he_wh, + heat_pump_time=hp_time, + heat_element_time=he_time, + heat_pump_percentage=hp_pct, + heat_element_percentage=he_pct, + ) + ) + return EnergyReport( + totals=_build_totals(energy_response.total), months=months + ) + + +def build_daily_energy_report( + energy_response: Any, year: int, month: int +) -> DailyEnergyReport | None: + """Aggregate a single month's daily energy into neutral rows. + + Args: + 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. + """ + month_data = energy_response.get_month_data(year, month) + if not month_data or not month_data.data: + return None + + days: list[EnergyPeriodRow] = [] + for day_num, day_data in enumerate(month_data.data, start=1): + total_wh = day_data.total_usage + hp_wh = day_data.heat_pump_usage + he_wh = day_data.heat_element_usage + hp_pct, he_pct = _percentages(hp_wh, he_wh, total_wh) + days.append( + EnergyPeriodRow( + label=str(day_num), + total_wh=total_wh, + heat_pump_wh=hp_wh, + heat_element_wh=he_wh, + heat_pump_time=day_data.heat_pump_time, + heat_element_time=day_data.heat_element_time, + heat_pump_percentage=hp_pct, + heat_element_percentage=he_pct, + ) + ) + return DailyEnergyReport( + year=year, + month=month, + totals=_build_totals(energy_response.total), + days=days, + ) diff --git a/src/nwp500/cli/rich_output.py b/src/nwp500/cli/rich_output.py index fed9d21c..72794f9d 100644 --- a/src/nwp500/cli/rich_output.py +++ b/src/nwp500/cli/rich_output.py @@ -1,57 +1,34 @@ -"""Rich-enhanced output formatting with graceful fallback.""" +"""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 :mod:`.presentation` +(and JSON/tables) using Rich exclusively. +""" import itertools import json import logging -import os from collections.abc import Callable -from typing import TYPE_CHECKING, Any, cast - -if TYPE_CHECKING: - from rich.console import Console - from rich.markdown import Markdown - from rich.panel import Panel - from rich.syntax import Syntax - from rich.table import Table - from rich.text import Text - from rich.tree import Tree - -_rich_available = False - -try: - from rich.console import Console - from rich.markdown import Markdown - from rich.panel import Panel - from rich.syntax import Syntax - from rich.table import Table - from rich.text import Text - from rich.tree import Tree - - _rich_available = True -except ImportError: - Console = None # type: ignore[assignment,misc] - Markdown = None # type: ignore[assignment,misc] - Panel = None # type: ignore[assignment,misc] - Syntax = None # type: ignore[assignment,misc] - Table = None # type: ignore[assignment,misc] - Text = None # type: ignore[assignment,misc] - Tree = None # type: ignore[assignment,misc] +from typing import Any, cast + +from rich.console import Console +from rich.markdown import Markdown +from rich.panel import Panel +from rich.syntax import Syntax +from rich.table import Table +from rich.text import Text +from rich.tree import Tree + +from .presentation import ( + DailyEnergyReport, + EnergyPeriodRow, + EnergyReport, + EnergyTotals, +) _logger = logging.getLogger(__name__) -def _should_use_rich() -> bool: - """Check if Rich should be used. - - Returns: - True if Rich is available and enabled, False otherwise. - """ - if not _rich_available: - return False - # Allow explicit override via environment variable - return os.getenv("NWP500_NO_RICH", "0") != "1" - - _MONTH_ABBR = [ "", "Jan", @@ -165,22 +142,14 @@ def _collapse_ranges( class OutputFormatter: - """Unified output formatter with Rich enhancement support. + """Rich output formatter for CLI human-readable output. - Automatically detects Rich availability and routes output to the - appropriate formatter. Falls back to plain text when Rich is - unavailable or explicitly disabled. + Rich is mandatory; this formatter always renders with Rich. """ def __init__(self) -> None: """Initialize the formatter.""" - self.use_rich = _should_use_rich() - self.console: Any - if self.use_rich: - assert Console is not None - self.console = Console() - else: - self.console = None + self.console = Console() def print_status_table(self, items: list[tuple[str, str, str]]) -> None: """Print status items as a formatted table. @@ -188,36 +157,36 @@ def print_status_table(self, items: list[tuple[str, str, str]]) -> None: Args: items: List of (category, label, value) tuples """ - if not self.use_rich: - self._print_status_plain(items) - else: - self._print_status_rich(items) + self._print_status_rich(items) - def print_energy_table(self, months: list[dict[str, Any]]) -> None: - """Print energy usage data as a formatted table. + def print_energy_table(self, report: EnergyReport) -> None: + """Print an energy usage report (summary + monthly breakdown). Args: - months: List of monthly energy data dictionaries + report: Neutral energy report from :mod:`.presentation` """ - if not self.use_rich: - self._print_energy_plain(months) - else: - self._print_energy_rich(months) + self._print_energy_summary(report.totals, "ENERGY USAGE REPORT") + if report.months: + self._print_energy_rich(report.months) - def print_daily_energy_table( - self, days: list[dict[str, Any]], year: int, month: int - ) -> None: - """Print daily energy usage data as a formatted table. + def print_daily_energy_table(self, report: DailyEnergyReport) -> None: + """Print a daily energy usage report (summary + daily breakdown). Args: - days: List of daily energy data dictionaries - year: Year for the data - month: Month for the data + report: Neutral daily energy report from :mod:`.presentation` """ - if not self.use_rich: - self._print_daily_energy_plain(days, year, month) - else: - self._print_daily_energy_rich(days, year, month) + from calendar import month_name + + year, month = report.year, report.month + month_str = ( + f"{month_name[month]} {year}" + if 1 <= month <= 12 + else f"Month {month} {year}" + ) + self._print_energy_summary( + report.totals, f"DAILY ENERGY USAGE - {month_str}" + ) + self._print_daily_energy_rich(report.days) def print_error( self, @@ -232,10 +201,7 @@ def print_error( title: Panel title details: Optional list of detail lines """ - if not self.use_rich: - self._print_error_plain(message, title, details) - else: - self._print_error_rich(message, title, details) + self._print_error_rich(message, title, details) def print_success(self, message: str) -> None: """Print a success message. @@ -243,10 +209,7 @@ def print_success(self, message: str) -> None: Args: message: Success message to display """ - if not self.use_rich: - print(f"✓ {message}") - else: - self._print_success_rich(message) + self._print_success_rich(message) def print_info(self, message: str) -> None: """Print an info message. @@ -254,10 +217,7 @@ def print_info(self, message: str) -> None: Args: message: Info message to display """ - if not self.use_rich: - print(f"ℹ {message}") - else: - self._print_info_rich(message) + self._print_info_rich(message) def print_device_list(self, devices: list[dict[str, Any]]) -> None: """Print list of devices with status indicators. @@ -265,10 +225,7 @@ def print_device_list(self, devices: list[dict[str, Any]]) -> None: Args: devices: List of device dictionaries with status info """ - if not self.use_rich: - self._print_device_list_plain(devices) - else: - self._print_device_list_rich(devices) + self._print_device_list_rich(devices) def print_tou_schedule( self, @@ -291,26 +248,15 @@ def print_tou_schedule( decode_week: Function to decode week bitfield decode_price_fn: Function to decode price values """ - if not self.use_rich: - self._print_tou_plain( - name, - utility, - zip_code, - schedules, - decode_season, - decode_week, - decode_price_fn, - ) - else: - self._print_tou_rich( - name, - utility, - zip_code, - schedules, - decode_season, - decode_week, - decode_price_fn, - ) + self._print_tou_rich( + name, + utility, + zip_code, + schedules, + decode_season, + decode_week, + decode_price_fn, + ) def print_reservations_table( self, reservations: list[dict[str, Any]], enabled: bool = False @@ -321,159 +267,11 @@ def print_reservations_table( reservations: List of reservation dictionaries enabled: Whether reservations are enabled globally """ - if not self.use_rich: - self._print_reservations_plain(reservations, enabled) - else: - self._print_reservations_rich(reservations, enabled) - - # Plain text implementations (fallback) - - def _print_status_plain(self, items: list[tuple[str, str, str]]) -> None: - """Plain text status output (fallback).""" - # Calculate widths - max_label = max((len(label) for _, label, _ in items), default=20) - max_value = max((len(str(value)) for _, _, value in items), default=20) - width = max_label + max_value + 4 - - # Print header - print("=" * width) - print("DEVICE STATUS") - print("=" * width) - - # Print items grouped by category - if items: - current_category: str | None = None - for category, label, value in items: - if category != current_category: - if current_category is not None: - print() - print(category) - print("-" * width) - current_category = category - print(f" {label:<{max_label}} {value}") - - print("=" * width) - - def _print_energy_plain(self, months: list[dict[str, Any]]) -> None: - """Plain text energy output (fallback).""" - # This is a simplified version - the actual rendering comes from - # output_formatters.format_energy_usage() - print("ENERGY USAGE REPORT") - print("=" * 90) - for month in months: - print(f"{month}") - - def _print_device_list_plain(self, devices: list[dict[str, Any]]) -> None: - """Plain text device list output (fallback).""" - if not devices: - print("No devices found") - return - - print("DEVICES") - print("-" * 80) - for device in devices: - name = device.get("name", "Unknown") - status = device.get("status", "Unknown") - temp = device.get("temperature", "N/A") - print(f" {name:<20} {status:<15} {temp}") - print("-" * 80) - - def _print_tou_plain( - self, - name: str, - utility: str, - zip_code: int, - schedules: Any, - decode_season: Any, - decode_week: Any, - decode_price_fn: Any, - ) -> None: - """Plain text TOU schedule output.""" - print("TOU SCHEDULE") - print("=" * 72) - print(f" Plan: {name}") - print(f" Utility: {utility}") - print(f" ZIP: {zip_code}") - print("=" * 72) - - for sched in schedules: - months = decode_season(sched.season) - month_str = _format_months(months) - print(f"\n Season: {month_str}") - print( - f" {'Days':<20} {'Time':>13}" - f" {'Min $/kWh':>10} {'Max $/kWh':>10}" - ) - print(f" {'-' * 57}") - for iv in sched.intervals: - days = decode_week(iv.get("week", 0)) - dp = iv.get("decimalPoint", 5) - p_min = decode_price_fn(iv.get("priceMin", 0), dp) - p_max = decode_price_fn(iv.get("priceMax", 0), dp) - time_str = ( - f"{iv.get('startHour', 0):02d}:" - f"{iv.get('startMinute', 0):02d}" - f"–{iv.get('endHour', 0):02d}:" - f"{iv.get('endMinute', 0):02d}" - ) - day_str = _abbreviate_days(days) - print( - f" {day_str:<20} {time_str:>13}" - f" {p_min:>10.5f} {p_max:>10.5f}" - ) - - def _print_reservations_plain( - self, reservations: list[dict[str, Any]], enabled: bool = False - ) -> None: - """Plain text reservations output (fallback).""" - status_str = "ENABLED" if enabled else "DISABLED" - print(f"Reservations: {status_str}") - print() - - if not reservations: - print("No reservations configured") - return - - print("RESERVATIONS") - print("=" * 110) - print( - f" {'#':<3} {'Enabled':<10} {'Days':<25} " - f"{'Time':<8} {'Mode':<20} {'Temp':<10}" - ) - print("=" * 110) - - for res in reservations: - num = res.get("number", "?") - is_enabled = res.get("enabled", False) - enabled_str = "Yes" if is_enabled else "No" - days_str = _abbreviate_days(res.get("days", [])) - time_str = res.get("time", "??:??") - mode = res.get("mode", "?") - temp = res.get("temperature", "?") - unit = res.get("unit", "") - temp_str = f"{temp}{unit}" if temp != "?" else "?" - print( - f" {num:<3} {enabled_str:<10} {days_str:<25} " - f"{time_str:<8} {mode:<20} {temp_str:<10}" - ) - print("=" * 110) - - def _print_error_plain( - self, - message: str, - title: str, - details: list[str] | None = None, - ) -> None: - """Plain text error output (fallback).""" - print(f"{title}: {message}") - if details: - for detail in details: - print(f" • {detail}") + self._print_reservations_rich(reservations, enabled) def _print_success_rich(self, message: str) -> None: """Rich-enhanced success output.""" assert self.console is not None - assert _rich_available panel = cast(Any, Panel)( f"[green]✓ {message}[/green]", border_style="green", @@ -484,7 +282,6 @@ def _print_success_rich(self, message: str) -> None: def _print_info_rich(self, message: str) -> None: """Rich-enhanced info output.""" assert self.console is not None - assert _rich_available panel = cast(Any, Panel)( f"[blue]ℹ {message}[/blue]", border_style="blue", @@ -495,7 +292,6 @@ def _print_info_rich(self, message: str) -> None: def _print_device_list_rich(self, devices: list[dict[str, Any]]) -> None: """Rich-enhanced device list output.""" assert self.console is not None - assert _rich_available if not devices: panel = cast(Any, Panel)("No devices found", border_style="yellow") @@ -544,7 +340,6 @@ def _print_tou_rich( ) -> None: """Rich-enhanced TOU schedule output.""" assert self.console is not None - assert _rich_available self.console.print() self.console.print( @@ -605,7 +400,6 @@ def _print_reservations_rich( ) -> None: """Rich-enhanced reservations output.""" assert self.console is not None - assert _rich_available status_color = "green" if enabled else "red" status_text = "ENABLED" if enabled else "DISABLED" @@ -652,14 +446,16 @@ def _print_reservations_rich( def _print_status_rich(self, items: list[tuple[str, str, str]]) -> None: """Rich-enhanced status output.""" assert self.console is not None - assert _rich_available table = cast(Any, Table)(title="DEVICE STATUS", show_header=False) if not items: - # If no items, just print the header using plain text - # to match expected output - self._print_status_plain(items) + # Preserve the previous empty-status header rendering. + width = 44 + print("=" * width) + print("DEVICE STATUS") + print("=" * width) + print("=" * width) return current_category: str | None = None @@ -681,12 +477,41 @@ def _print_status_rich(self, items: list[tuple[str, str, str]]) -> None: self.console.print(table) - def _print_energy_rich(self, months: list[dict[str, Any]]) -> None: - """Rich-enhanced energy output.""" + def _print_energy_summary(self, totals: EnergyTotals, title: str) -> None: + """Render the shared energy 'TOTAL SUMMARY' block as a Rich table.""" assert self.console is not None - assert _rich_available - table = cast(Any, Table)(title="ENERGY USAGE REPORT", show_header=True) + table = cast(Any, Table)(title=title, show_header=False) + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green", justify="right") + + total_kwh = totals.total_usage_wh / 1000 + table.add_row( + "Total Energy Used", + f"{totals.total_usage_wh:,} Wh ({total_kwh:.2f} kWh)", + ) + table.add_row( + " Heat Pump", + f"{totals.heat_pump_usage_wh:,} Wh " + f"({totals.heat_pump_percentage:.1f}%)", + ) + table.add_row( + " Heat Element", + f"{totals.heat_element_usage_wh:,} Wh " + f"({totals.heat_element_percentage:.1f}%)", + ) + table.add_row("Total Time Running", f"{totals.total_time_hours} hours") + table.add_row(" Heat Pump", f"{totals.heat_pump_time_hours} hours") + table.add_row( + " Heat Element", f"{totals.heat_element_time_hours} hours" + ) + self.console.print(table) + + def _print_energy_rich(self, months: list[EnergyPeriodRow]) -> None: + """Rich-enhanced monthly energy breakdown.""" + assert self.console is not None + + table = cast(Any, Table)(title="MONTHLY BREAKDOWN", show_header=True) table.add_column("Month", style="cyan", width=15) table.add_column( "Total kWh", style="magenta", justify="right", width=12 @@ -695,42 +520,39 @@ def _print_energy_rich(self, months: list[dict[str, Any]]) -> None: table.add_column("HE Usage", width=18) for month in months: - month_str = month.get("month_str", "N/A") - total_kwh = month.get("total_kwh", 0) - hp_kwh = month.get("hp_kwh", 0) - he_kwh = month.get("he_kwh", 0) - hp_pct = month.get("hp_pct", 0) - he_pct = month.get("he_pct", 0) - - # Create progress bar representations - hp_bar = self._create_progress_bar(hp_pct, 10) - he_bar = self._create_progress_bar(he_pct, 10) - - # Color code based on efficiency - hp_color = ( - "green" - if hp_pct >= 70 - else ("yellow" if hp_pct >= 50 else "red") - ) - he_color = ( - "red" - if he_pct >= 50 - else ("yellow" if he_pct >= 30 else "green") - ) - - hp_text = ( - f"{hp_kwh:.1f} kWh " - f"[{hp_color}]{hp_pct:.0f}%[/{hp_color}]\n{hp_bar}" - ) - he_text = ( - f"{he_kwh:.1f} kWh " - f"[{he_color}]{he_pct:.0f}%[/{he_color}]\n{he_bar}" + total_kwh = month.total_wh / 1000 + hp_kwh = month.heat_pump_wh / 1000 + he_kwh = month.heat_element_wh / 1000 + hp_pct = month.heat_pump_percentage + he_pct = month.heat_element_percentage + + hp_text, he_text = self._energy_usage_cells( + hp_kwh, hp_pct, he_kwh, he_pct ) - - table.add_row(month_str, f"{total_kwh:.1f}", hp_text, he_text) + table.add_row(month.label, f"{total_kwh:.1f}", hp_text, he_text) self.console.print(table) + def _energy_usage_cells( + self, hp_kwh: float, hp_pct: float, he_kwh: float, he_pct: float + ) -> tuple[str, str]: + """Build the HP/HE usage cell markup shared by energy tables.""" + hp_bar = self._create_progress_bar(hp_pct, 10) + he_bar = self._create_progress_bar(he_pct, 10) + hp_color = ( + "green" if hp_pct >= 70 else ("yellow" if hp_pct >= 50 else "red") + ) + he_color = ( + "red" if he_pct >= 50 else ("yellow" if he_pct >= 30 else "green") + ) + hp_text = ( + f"{hp_kwh:.1f} kWh [{hp_color}]{hp_pct:.0f}%[/{hp_color}]\n{hp_bar}" + ) + he_text = ( + f"{he_kwh:.1f} kWh [{he_color}]{he_pct:.0f}%[/{he_color}]\n{he_bar}" + ) + return hp_text, he_text + def _create_progress_bar(self, percentage: float, width: int = 10) -> str: """Create a simple progress bar string. @@ -745,41 +567,11 @@ def _create_progress_bar(self, percentage: float, width: int = 10) -> str: bar = "█" * filled + "░" * (width - filled) return f"[{bar}]" - def _print_daily_energy_plain( - self, days: list[dict[str, Any]], year: int, month: int - ) -> None: - """Plain text daily energy output (fallback).""" - # This is a simplified version - the actual rendering comes from - # output_formatters.format_daily_energy_usage() - from calendar import month_name - - month_str = ( - f"{month_name[month]} {year}" - if 1 <= month <= 12 - else f"Month {month} {year}" - ) - print(f"DAILY ENERGY USAGE - {month_str}") - print("=" * 100) - for day in days: - print(f"{day}") - - def _print_daily_energy_rich( - self, days: list[dict[str, Any]], year: int, month: int - ) -> None: - """Rich-enhanced daily energy output.""" - from calendar import month_name - + def _print_daily_energy_rich(self, days: list[EnergyPeriodRow]) -> None: + """Rich-enhanced daily energy breakdown.""" assert self.console is not None - assert _rich_available - month_str = ( - f"{month_name[month]} {year}" - if 1 <= month <= 12 - else f"Month {month} {year}" - ) - table = cast(Any, Table)( - title=f"DAILY ENERGY USAGE - {month_str}", show_header=True - ) + table = cast(Any, Table)(title="DAILY BREAKDOWN", show_header=True) table.add_column("Day", style="cyan", width=6) table.add_column( "Total kWh", style="magenta", justify="right", width=12 @@ -788,39 +580,16 @@ def _print_daily_energy_rich( table.add_column("HE Usage", width=18) for day in days: - day_num = day.get("day", "N/A") - total_kwh = day.get("total_kwh", 0) - hp_kwh = day.get("hp_kwh", 0) - he_kwh = day.get("he_kwh", 0) - hp_pct = day.get("hp_pct", 0) - he_pct = day.get("he_pct", 0) - - # Create progress bar representations - hp_bar = self._create_progress_bar(hp_pct, 10) - he_bar = self._create_progress_bar(he_pct, 10) - - # Color code based on efficiency - hp_color = ( - "green" - if hp_pct >= 70 - else ("yellow" if hp_pct >= 50 else "red") - ) - he_color = ( - "red" - if he_pct >= 50 - else ("yellow" if he_pct >= 30 else "green") + total_kwh = day.total_wh / 1000 + hp_kwh = day.heat_pump_wh / 1000 + he_kwh = day.heat_element_wh / 1000 + hp_text, he_text = self._energy_usage_cells( + hp_kwh, + day.heat_pump_percentage, + he_kwh, + day.heat_element_percentage, ) - - hp_text = ( - f"{hp_kwh:.1f} kWh " - f"[{hp_color}]{hp_pct:.0f}%[/{hp_color}]\n{hp_bar}" - ) - he_text = ( - f"{he_kwh:.1f} kWh " - f"[{he_color}]{he_pct:.0f}%[/{he_color}]\n{he_bar}" - ) - - table.add_row(str(day_num), f"{total_kwh:.1f}", hp_text, he_text) + table.add_row(day.label, f"{total_kwh:.1f}", hp_text, he_text) self.console.print(table) @@ -832,7 +601,6 @@ def _print_error_rich( ) -> None: """Rich-enhanced error output.""" assert self.console is not None - assert _rich_available content = f"❌ {title}\n\n{message}" if details: @@ -855,10 +623,7 @@ def print_json_highlighted(self, data: Any) -> None: Args: data: Data to print as JSON """ - if not self.use_rich: - print(json.dumps(data, indent=2, default=str)) - else: - self._print_json_highlighted_rich(data) + self._print_json_highlighted_rich(data) def print_device_tree( self, device_name: str, device_info: dict[str, Any] @@ -869,10 +634,7 @@ def print_device_tree( device_name: Name of the device device_info: Dictionary of device information """ - if not self.use_rich: - self._print_device_tree_plain(device_name, device_info) - else: - self._print_device_tree_rich(device_name, device_info) + self._print_device_tree_rich(device_name, device_info) def print_markdown_report(self, markdown_content: str) -> None: """Print markdown-formatted content. @@ -880,31 +642,13 @@ def print_markdown_report(self, markdown_content: str) -> None: Args: markdown_content: Markdown formatted string """ - if not self.use_rich: - print(markdown_content) - else: - self._print_markdown_rich(markdown_content) - - # Plain text implementations (Phase 3 fallback) - - def _print_json_highlighted_plain(self, data: Any) -> None: - """Plain text JSON output (fallback).""" - print(json.dumps(data, indent=2, default=str)) - - def _print_device_tree_plain( - self, device_name: str, device_info: dict[str, Any] - ) -> None: - """Plain text tree output (fallback).""" - print(f"Device: {device_name}") - for key, value in device_info.items(): - print(f" {key}: {value}") + self._print_markdown_rich(markdown_content) # Rich implementations (Phase 3) def _print_json_highlighted_rich(self, data: Any) -> None: """Rich-enhanced JSON output with syntax highlighting.""" assert self.console is not None - assert _rich_available json_str = json.dumps(data, indent=2, default=str) syntax = cast(Any, Syntax)( @@ -917,7 +661,6 @@ def _print_device_tree_rich( ) -> None: """Rich-enhanced tree output for device information.""" assert self.console is not None - assert _rich_available tree = cast(Any, Tree)(f"📱 {device_name}", guide_style="bold cyan") @@ -960,7 +703,6 @@ def _print_device_tree_rich( def _print_markdown_rich(self, content: str) -> None: """Rich-enhanced markdown rendering.""" assert self.console is not None - assert _rich_available markdown = cast(Any, Markdown)(content) self.console.print(markdown)