"""Command line tools to interact with the Insteon devices.""" from .. import devices from ..constants import ALDBStatus, DeviceCategory from ..managers.link_manager import async_link_devices from ..utils import seconds_to_ramp_rate from .tools_base import ToolsBase SHOW_ADVANCED = True class ToolsAldb(ToolsBase): """Command class to test interactivity.""" async def do_load_aldb( self, address, clear=None, log_stdout=None, background=False ): """Load the All-Link Database of a device. Usage: load_aldb [--background | -b] address [clear] address: Insteon address of a device or `all` for all devices clear: Clear the current ALDB records from memory to ensure a full DB read y | n Default n """ ask_clear = address is None and not background try: addresses = await self._ensure_address( address=address, name="Address", ask_value=not background, log_stdout=log_stdout, allow_all=True, match_device=True, ) if not addresses: return except ValueError: log_stdout("Invalid device address or device not found") return try: refresh = await self._ensure_bool( clear, "Clear current records", ask_clear, log_stdout, default=False ) except ValueError: log_stdout("An invalid value for `clear` was received") return battery_devices = [] for device_address in addresses: if devices[device_address].is_battery: battery_devices.append(device_address) # Only load the modem ALDB if explicitly asked if devices[device_address] == devices.modem and len(addresses) == 1: await devices.modem.aldb.async_load() elif devices[device_address].cat == 0x03: pass else: # tasks.append(devices[device_address].aldb.async_load(refresh=refresh)) await devices[device_address].aldb.async_load(refresh=refresh) if battery_devices: log_stdout("The following devices are battery operated.") log_stdout("They will load in the background when they wake up.") for device_address in battery_devices: log_stdout(f" - {device_address}") # if the device did not load the first time, try one more time with refresh for device_address in addresses: if ( devices[device_address] != devices.modem and devices[device_address].aldb.status != ALDBStatus.LOADED and not devices[device_address].is_battery ): await devices[device_address].aldb.async_load(refresh=True) async def do_print_aldb( self, address, unused=None, log_stdout=None, background=False ): """Print the records in an All-Link Database. Usage: print_aldb [--background | -b] address | all unused address: Insteon address of a device or `all` for all devices unused (optional): y | n """ try: addresses = await self._ensure_address( address=address, name="Address", ask_value=not background, log_stdout=log_stdout, allow_all=False, match_device=True, ) if not addresses: return except ValueError: log_stdout("Invalid device address or device not found") return try: unused = await self._ensure_bool( unused, "Show Unused", not background, log_stdout, default=False ) except ValueError: unused = False self._print_aldb_out(addresses, unused, log_stdout) async def do_add_default_links(self, address, log_stdout=None, background=False): """Add default links between a device and the modem. Usage: add_default_links [--background | -b] address address: Insteon address of a device """ try: addresses = await self._ensure_address( address=address, name="Address", ask_value=not background, log_stdout=log_stdout, allow_all=False, match_device=True, ) if not addresses: return address = addresses[0] except ValueError: log_stdout("Invalid device address or device not found") return device = devices[address] await device.async_add_default_links() async def do_add_device_to_scene( self, address, scene, data1=None, data2=None, data3=None, log_stdout=None, background=False, ): """Add a device to a scene. Usage: add_device_to_scene [--background | -b]
[] [] [