Example using the SmartDeviceView
Below is a example application using the SmartDeviceView class. This class provides a read-only, thread safe copy of the device and component status information. It's typically used in conjunction with the SmartDeviceWorker class.
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
"""Basic example of SmartDevice control."""
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 switch
# device_identity = "Sydney Dev A"
# output_identity = "Sydney Dev A O1"
# meter_identity = "Sydney Dev A M1"
# Test a Tasmota switch
device_identity = "Sydney Dev B"
output_identity = "Sydney Dev B O1"
meter_identity = "Sydney Dev B M1"
def test_basic(logger: SCLogger, smart_switch_control: SCSmartDevice) -> None:
"""Test function for basic SmartSwitch control."""
logger.log_message(f"\n\n\nTesting basic functionality for device: {device_identity}", "summary")
# Get the device
try:
device = smart_switch_control.get_device(device_identity)
device_status = smart_switch_control.get_device_status(device)
if device_status:
logger.log_message(f"Device {device_identity} is online.", "summary")
else:
logger.log_message(f"Device {device_identity} is offline or not found.", "summary")
except RuntimeError as e:
logger.log_message(f"Error getting status for device {device_identity}: {e}", "error")
sys.exit(1)
except TimeoutError as e:
logger.log_message(f"Timeout error getting status for device {device_identity}: {e}", "error")
else:
logger.log_message(f"{device_identity} before output change:\n {smart_switch_control.print_device_status(device_identity)}", "detailed")
# Get the output component and its current state
output_obj = smart_switch_control.get_device_component("output", output_identity)
is_online = smart_switch_control.is_device_online(device)
smart_switch_control.get_device_status(device)
current_state = output_obj.get("State", False) # Default to False if State is not found
logger.log_message(f"#1 Output status for {output_identity}: Is Online: {is_online}, Current State: {current_state}", "detailed")
# Change the output state to the opposite of the current state
logger.log_message("Waiting 3 seconds before changing the output state...", "detailed")
for i in range(3):
time.sleep(1) # Short delay to ensure the device is ready for the next command
print(f"{i + 1}...", end="", flush=True)
logger.log_message("Attempting to change the output state...", "detailed")
smart_switch_control.change_output(output_identity, not current_state)
# Get the meter reading and output status again after the change
meter_obj = smart_switch_control.get_device_component("meter", meter_identity)
meter_reading = meter_obj.get("Energy", None)
logger.log_message(f"Meter reading for {meter_identity}: {meter_reading}", "detailed")
# Get the latest device status to check if the output change was successful
is_online = smart_switch_control.is_device_online(device)
smart_switch_control.get_device_status(device)
current_state = output_obj.get("State", False) # Default to False if State is not found
logger.log_message(f"#2 Output status for {output_identity}: Is Online: {is_online}, Current State: {current_state}", "detailed")
# Log the full device status after the change
logger.log_message(f"{device_identity} after output change:\n {smart_switch_control.print_device_status(device_identity)}", "detailed")
print(smart_switch_control.print_device_status(device_identity))
def main():
"""Main function to run the example code."""
print(f"Hello from switch_basic running on {platform.system()}")
# Initialize the configuration manager, logger, and Smart_switch 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_basic(logger, smart_switch_control)
if __name__ == "__main__":
main()