python-kasa · rytilahti · May 10, 2021 · Mar 18, 2021 · Mar 18, 2021 · May 10, 2021
diff --git a/devtools/dump_devinfo.py b/devtools/dump_devinfo.py
@@ -0,0 +1,141 @@
+"""This script generates devinfo files for the test suite.
+
+If you have new, yet unsupported device or a device with no devinfo file under kasa/tests/fixtures,
+feel free to run this script and create a PR to add the file to the repository.
+
+Executing this script will several modules and methods one by one,
+and finally execute a query to query all of them at once.
+"""
+import asyncio
+import collections.abc
+import json
+import logging
+import re
+from collections import defaultdict, namedtuple
+from pprint import pprint
+
+import click
+
+from kasa import TPLinkSmartHomeProtocol
+
+Call = namedtuple("Call", "module method")
+
+
+def scrub(res):
+    """Remove identifiers from the given dict."""
+    keys_to_scrub = [
+        "deviceId",
+        "fwId",
+        "hwId",
+        "oemId",
+        "mac",
+        "mic_mac",
+        "latitude_i",
+        "longitude_i",
+    <
10000
span class='blob-code-inner blob-code-marker ' data-code-marker="+">        "latitude",
+        "longitude",
+    ]
+
+    for k, v in res.items():
+        if isinstance(v, collections.abc.Mapping):
+            res[k] = scrub(res.get(k))
+        else:
+            if k in keys_to_scrub:
+                if k in ["latitude_i", "longitude_i"]:
+                    v = 0
+                else:
+                    v = re.sub(r"\w", "0", v)
+
+            res[k] = v
+    return res
+
+
+def default_to_regular(d):
+    """Convert nested defaultdicts to regular ones.
+
+    From https://stackoverflow.com/a/26496899
+    """
+    if isinstance(d, defaultdict):
+        d = {k: default_to_regular(v) for k, v in d.items()}
+    return d
+
+
+@click.command()
+@click.argument("host")
+@click.option("--debug")
+def cli(host, debug):
+    """Generate devinfo file for given device."""
+    if debug:
+        logging.basicConfig(level=logging.DEBUG)
+
+    items = [
+        Call(module="system", method="get_sysinfo"),
+        Call(module="emeter", method="get_realtime"),
+        Call(module="smartlife.iot.dimmer", method="get_dimmer_parameters"),
+        Call(module="smartlife.iot.common.emeter", method="get_realtime"),
+        Call(
+            module="smartlife.iot.smartbulb.lightingservice", method="get_light_state"
+        ),
+    ]
+
+    protocol = TPLinkSmartHomeProtocol()
+
+    successes = []
+
+    for test_call in items:
+        try:
+            click.echo(f"Testing {test_call}..", nl=False)
+            info = asyncio.run(
+                protocol.query(host, {test_call.module: {test_call.method: None}})
+            )
+            resp = info[test_call.module]
+        except Exception as ex:
+            click.echo(click.style(f"FAIL {ex}", fg="red"))
+        else:
+            if "err_msg" in resp:
+                click.echo(click.style(f"FAIL {resp['err_msg']}", fg="red"))
+            else:
+                click.echo(click.style("OK", fg="green"))
+                successes.append((test_call, info))
+
+    final_query = defaultdict(defaultdict)
+    final = defaultdict(defaultdict)
+
+    for succ, resp in successes:
+        final_query[succ.module][succ.method] = None
+        final[succ.module][succ.method] = resp
+
+    final = default_to_regular(final)
+
+    try:
+        final = asyncio.run(protocol.query(host, final_query))
+    except Exception as ex:
+        click.echo(
+            click.style(
+                f"Unable to query all successes at once: {ex}", bold=True, fg="red"
+            )
+        )
+
+    click.echo("Got %s successes" % len(successes))
+    click.echo(click.style("## device info file ##", bold=True))
+
+    sysinfo = final["system"]["get_sysinfo"]
+    model = sysinfo["model"]
+    hw_version = sysinfo["hw_ver"]
+    sw_version = sysinfo["sw_ver"]
+    sw_version = sw_version.split(" ", maxsplit=1)[0]
+    save_to = f"{model}_{hw_version}_{sw_version}.json"
+    pprint(scrub(final))
+    save = click.prompt(f"Do you want to save the above content to {save_to} (y/n)")
+    if save == "y":
+        click.echo(f"Saving info to {save_to}")
+
+        with open(save_to, "w") as f:
+            json.dump(final, f, sort_keys=True, indent=4)
+            f.write("\n")
+    else:
+        click.echo("Not saving.")
+
+
+if __name__ == "__main__":
+    cli()
diff --git a/kasa/cli.py b/kasa/cli.py
@@ -1,7 +1,5 @@
 """python-kasa cli tool."""
-import json
 import logging
-import re
 from pprint import pformat as pf
 from typing import cast
 
@@ -131,39 +129,9 @@ async def dump_discover(ctx, scrub):
 
     Useful for dumping into a file to be added to the test suite.
     """
-    target = ctx.parent.params["target"]
-    keys_to_scrub = [
-        "deviceId",
-        "fwId",
-        "hwId",
-        "oemId",
-        "mac",
-        "latitude_i",
-        "longitude_i",
-        "latitude",
-        "longitude",
-    ]
-    devs = await Discover.discover(target=target, return_raw=True)
-    if scrub:
-        click.echo("Scrubbing personal data before writing")
-    for dev in devs.values():
-        if scrub:
-            for key in keys_to_scrub:
-                if key in dev["system"]["get_sysinfo"]:
-                    val = dev["system"]["get_sysinfo"][key]
-                    if key in ["latitude_i", "longitude_i"]:
-                        val = 0
-                    else:
-                        val = re.sub(r"\w", "0", val)
-                    dev["system"]["get_sysinfo"][key] = val
-
-        model = dev["system"]["get_sysinfo"]["model"]
-        hw_version = dev["system"]["get_sysinfo"]["hw_ver"]
-        save_to = f"{model}_{hw_version}.json"
-        click.echo(f"Saving info to {save_to}")
-        with open(save_to, "w") as f:
-            json.dump(dev, f, sort_keys=True, indent=4)
-            f.write("\n")
+    click.echo(
+        "This is deprecated, use the script inside devtools to generate a devinfo file."
+    )
 
 
 @cli.command()

diff --git a/kasa/discover.py b/kasa/discover.py
@@ -135,10 +135,6 @@ class Discover:
 
     DISCOVERY_QUERY = {
         "system": {"get_sysinfo": None},
-        "emeter": {"get_realtime": None},
-        "smartlife.iot.dimmer": {"get_dimmer_parameters": None},
-        "smartlife.iot.common.emeter": {"get_realtime": None},
-        "smartlife.iot.smartbulb.lightingservice": {"get_light_state": None},
     }
 
     @staticmethod