Support client.getDeviceList

[luqmanoop]

Hi @mihai-dinculescu, first of all I want to show my gratitude to you for creating this package. I've been able to use it do some really cool stuff at home.

Request: I was thinking if there's a way to do client.getDeviceList() after initializing the ApiClient without having to supply the address of the different devices to e.g. client.p100("IP_ADDRESS_OF_DEVICE")? This will enable programmatically getting the address of connected devices without having to supply it manually

Why? I have a cron job that periodically checks if a particular device is turned off to do some stuff but the problem I noticed after letting the cron run for about 10hrs is the script had crashed and the reason is because the IP address of that device had changed!

Device IP Address changes whenever router is rebooted

[mihai-dinculescu]

You can configure your router to statically assign an IP to your device as a quick workaround.

Nevertheless, discovering all the Tapo devices on your network and/or Tapo Cloud account is a great feature to have. I'm not sure when I'll be able to add it, but if someone wants to have a crack at it, I'm happy to help.

[luqmanoop]

@mihai-dinculescu Thanks. I'm unable to assign static IP with Starlink but I had a workaround!

Since each smart plug has a MAC address that is unchangeable but IP can be dynamic (due to restarting router). I noted down each smart plug MAC address & mapped them to corresponding IP output from arp-scan

The command I use for arp-scan is

sudo arp-scan -l -q --plain

This makes sure to only output just IP & MAC address of the connected devices to my network making sure I always get the latest assigned IP for each device!

e.g. output

192.168.1.1     0a:0c:0c:0a:0c:7d
192.168.1.25    0c:0c:0a:0f:06:85

I then converted the string output to json. I was able to do all of this in a python environment on my Raspberry Pi.

Pretty happy with the result

[alfadormx]

Hello, I am interested in this functionality too, something like Python-MagicHue is doing and that is being taking advantage of in Touch-Portal-MagicHome-Plugin.

I wonder too if it is necessary too to add the login and password to manipulate lights in a local network?

[nicklansley]

Hello - rather than pull this repo I'm just going to put my getDeviceList() implementation example here. It takes advantage of the fact we are in the asyncio world, so can take engage its timeout mechanism.

It simply blasts all IP addresses concurrently! See further down if you want to play nicer on your network...

My assumption is that IF a Tapo device is going to respond, it will do so in 1 second. In practice it is instant, so after 1 second we dump any async task that has not completed. Of course, you can increase this as needed if your Tapo devices respond sluggishly.

def get_useful_device_info(device_info):
    useful_info = {}
    useful_properties = ['avatar', 'device_on', 'model', 'nickname', 'signal_level', 'ssid']
    for property in dir(device_info):
        if property in useful_properties:
            useful_info[property] = getattr(device_info, property)

    return useful_info

async def device_probe(client, ip_address):
    device = await client.generic_device(ip_address)
    device_info = await device.get_device_info()
    if device_info:
        device_instance = {
            'ip_address': ip_address,
            'device_info': get_useful_device_info(device_info)
        }
        return True, device_instance
    return False, None

async def getDeviceList(client, timeout_seconds=1.0):
    device_data = []
    tasks = []

    for ip_octet in range(1, 253):
        ip_address = f"192.168.1.{ip_octet}"
        task = asyncio.create_task(asyncio.wait_for(device_probe(client, ip_address), timeout=timeout_seconds))
        tasks.append(task)

    for task in asyncio.as_completed(tasks):
        try:
            is_device, device_instance = await task
            if is_device:
                device_data.append(device_instance)
        except asyncio.TimeoutError:
            pass
        except Exception as e:
            pass

    return device_data

Call it like this:

async def run_async():
    client = tapo.ApiClient(tapo_email, tapo_password)
    # get all the client devices
    devices = await getDeviceList(client)
    print('Devices:', json.dumps(devices, indent=2))

if __name__ == "__main__":
    # Get email and password from environmental variables
    tapo_email = os.environ.get('TAPO_EMAIL')
    tapo_password = os.environ.get('TAPO_PASSWORD')
    asyncio.run(run_async())

My output (after a just a couple of seconds of starting the script):

  {
    "ip_address": "192.168.1.148",
    "device_info": {
      "avatar": "table_lamp",
      "device_on": false,
      "model": "P100",
      "nickname": "computer room lamp",
      "signal_level": 3,
      "ssid": "ssid-name"
    }
  },
  {
    "ip_address": "192.168.1.6",
    "device_info": {
      "avatar": "table_lamp",
      "device_on": false,
      "model": "P100",
      "nickname": "Nick Bedside Lamp",
      "signal_level": 3,
      "ssid": "ssid-name"
    }
  },
  {
    "ip_address": "192.168.1.59",
    "device_info": {
      "avatar": "kettle",
      "device_on": true,
      "model": "P100",
      "nickname": "Kettle",
      "signal_level": 2,
      "ssid": "ssid-name"
    }
  },
  {
    "ip_address": "192.168.1.129",
    "device_info": {
      "avatar": "table_lamp",
      "device_on": true,
      "model": "P100",
      "nickname": "Bernard Bedside Lamp",
      "signal_level": 3,
      "ssid": "ssid-name"
    }
  },
  {
    "ip_address": "192.168.1.137",
    "device_info": {
      "avatar": "table_lamp",
      "device_on": false,
      "model": "P100",
      "nickname": "Livingroom Fireside Lamp",
      "signal_level": 2,
      "ssid": "ssid-name"
    }
  }
]

If you want to play nice on your network, then you can use a Semaphore. Here are the same functions adjusted to only allow 10 concurrent tasks by default in the task list to run concurrently. Adjust the 'limit' property in getDeviceList() to be more passive / aggressive:

async def device_probe_semaphore(sem, client, ip_address, timeout_seconds):
    async with sem:
        return await device_probe(client, ip_address, timeout_seconds)

async def getDeviceList(client, limit=10, timeout_seconds=1.0):
    device_data = []
    sem = asyncio.Semaphore(limit)  # Limit concurrent tasks

    tasks = []
    for ip_octet in range(1, 253):
        ip_address = f"192.168.1.{ip_octet}"
        task = asyncio.create_task(device_probe_semaphore(sem, client, ip_address, timeout_seconds))
        tasks.append(task)

    for task in asyncio.as_completed(tasks):
        try:
            is_device, device_instance = await task
            if is_device:
                device_data.append(device_instance)
        except asyncio.TimeoutError:
            pass
        except Exception as e:
            pass
    return device_data

Weakness to overcome:

1. The IP address is assumed to be '192.168.1.x'. Solution here would be getting the computer's own IP address and using its first three octets when building the prospective device IP addresses.
2. It may be better to return the entire device object rather than 'useful' properties. The latter works for me but it's a personal choice unsuited for a public repo I suspect.