Example reading energy meters

Below is a example application that reads a device's energy meters

Module to initialise the SCSmartDevice instance

See the example config page for the YAML configuration used by this example.

  """Manual testing code for the ShellyControl class."""

  import threading

  from mergedeep import merge
  from sc_foundation import (
      SCCommon,
      SCConfigManager,
      SCLogger,
  )
  from validation_extras import smart_switch_extra_validation

  from sc_smart_device import SCSmartDevice, smart_devices_validator

  CONFIG_FILE = "examples/switch_config.yaml"


  def switch_init(wake_event: threading.Event | None = None, extra_validation: dict | None = None) -> tuple[SCConfigManager, SCLogger, SCSmartDevice]:
      """Create an instance of the SCConfigManager, SCLogger and SCSmartDevice class.

      Args:
          wake_event (threading.Event | None): Optional threading event to signal webhook events.
          extra_validation (dict | None): Optional extra validation schema to merge with the default schema.

      Returns:
          tuple[SCConfigManager, SCLogger, SCSmartDevice]: A tuple containing the initialized SCConfigManager, SCLogger, and SCSmartDevice instances.

      Raises:
          RuntimeError: If there is an error with the configuration file, logger initialization, or SCSmartDevice initialization.
      """
      # Merge the SmartDevices validation schema with the default validation schema
      merged_schema = merge({}, smart_devices_validator, smart_switch_extra_validation, extra_validation or {})
      assert isinstance(merged_schema, dict), "Merged schema should be type dict"

      # Initialize the SC_ConfigManager class
      try:
          config = SCConfigManager(
              config_file=CONFIG_FILE,
              validation_schema=merged_schema,
          )
      except RuntimeError as e:
          error_msg = f"Configuration file error: {e}"
          raise RuntimeError(error_msg) from e

      # Initialize the SC_Logger class
      try:
          logger_settings = config.get_logger_settings()
          logger = SCLogger(logger_settings)
      except RuntimeError as e:
          error_msg = f"Logger initialisation error: {e}"
          raise RuntimeError(error_msg) from e

      # Test internet connection
      if not SCCommon.check_internet_connection():
          logger.log_message("No internet connection detected.", "error")

      smart_switch_settings = config.get("SCSmartDevices")

      if smart_switch_settings is None:
          error_msg = "No SmartDevices settings found in the configuration file."
          raise RuntimeError(error_msg)

      # Initialize the SCSmartDevice class
      try:
          smart_switch_control = SCSmartDevice(logger, smart_switch_settings, wake_event)
      except RuntimeError as e:
          error_msg = f"SCSmartDevice initialization error: {e}"
          raise RuntimeError(error_msg) from e
      logger.log_message(f"SCSmartDevice initialized successfully with {len(smart_switch_control.devices)} devices.", "summary")

      return config, logger, smart_switch_control

Example application

  """Example of using the SmartDevice control to read energy meters."""
  import platform
  import sys
  import time

  from sc_foundation import SCLogger
  from switch_init import switch_init

  from sc_smart_device import SCSmartDevice

  # ------- Uncomment the relevant section below to test different devices and meters -------
  # Test a Shelly energy Meter
  meter_device_name = "Sydney Panel EM1"
  output_name = "Sydney Panel EM1 O1"
  meter1_name = "Sydney Panel EM1 M1"
  meter2_name = "Sydney Panel EM1 M2"

  # Test a Tasmota switch
  # meter_device_name = "Sydney Dev B"
  # output_name = "Sydney Dev B O1"
  # meter1_name = "Sydney Dev B M1"
  # meter2_name = None


  def test_new_meter(logger: SCLogger, smart_switch_control: SCSmartDevice):
      """Test SmartDevices energy meter functionality."""
      loop_delay = 1
      loop_count = 0
      max_loops = 5

      logger.log_message(f"Testing meter functionality for device: {meter_device_name}", "summary")

      # Get the device and components
      try:
          meter_device = smart_switch_control.get_device(meter_device_name)
          output = smart_switch_control.get_device_component("output", output_name)
          meter1 = smart_switch_control.get_device_component("meter", meter1_name)
          meter2 = smart_switch_control.get_device_component("meter", meter2_name) if meter2_name else None
      except RuntimeError as e:
          print(f"Error getting device: {e}", file=sys.stderr)
          sys.exit(1)
      except TimeoutError as e:
          print(f"Timeout error getting device: {e}", file=sys.stderr)
          sys.exit(1)
      else:
          logger.log_message(f"{meter_device_name} status:\n {smart_switch_control.print_device_status(meter_device_name)}", "detailed")

          while loop_count < max_loops:
              # Refresh the status of all devices
              smart_switch_control.get_device_status(meter_device)

              meter1_volts = meter1.get("Voltage", False)
              meter1_power = meter1.get("Power", False)
              meter1_energy = meter1.get("Energy", False)
              meter2_volts = meter2.get("Voltage", False) if meter2 else False
              meter2_power = meter2.get("Power", False) if meter2 else False
              meter2_energy = meter2.get("Energy", False) if meter2 else False

              logger.log_message(f"{output_name} State: {output.get('State', False)}.", "detailed")
              logger.log_message(f"{meter1_name} Volts: {meter1_volts}, Power: {meter1_power}, Energy: {meter1_energy}.", "detailed")
              if meter2:
                  logger.log_message(f"{meter2_name} Volts: {meter2_volts}, Power: {meter2_power}, Energy: {meter2_energy}.", "detailed")

              time.sleep(loop_delay)
              loop_count += 1


  def main():
      """Main function to run the example code."""
      print(f"Hello from switch_meter running on {platform.system()}")

      # Initialize the configuration manager, logger, and SmartDevices control
      try:
          _config, logger, smart_switch_control = switch_init()
      except RuntimeError as e:
          print(f"Initialization error: {e}", file=sys.stderr)
          sys.exit(1)

      test_new_meter(logger, smart_switch_control)


  if __name__ == "__main__":
      main()