Merge pull request #55 from mill1000/feature/new_cli

Implement a new CLI with support for querying device state and capabilities
mill1000 · Sep 14, 2023 · a5bcb02 · a5bcb02
2 parents eb285e4 + 35b3e3e
commit a5bcb02
Show file tree

Hide file tree

Showing 6 changed files with 263 additions and 59 deletions.
diff --git a/Dockerfile b/Dockerfile
@@ -12,4 +12,4 @@ FROM alpine:latest
 RUN apk add --update python3 py3-pip py3-pycryptodome
 COPY --from=build /msmart-build/dist/msmart_ng-*.whl /tmp
 RUN pip install /tmp/msmart_ng-*.whl
-ENTRYPOINT ["/usr/bin/midea-discover"]
+ENTRYPOINT ["/usr/bin/msmart-ng"]
diff --git a/README.md b/README.md
@@ -52,64 +52,94 @@ Some external dependencies have been replaced with standard Python modules.
 
 ## Installing
 Use pip, remove the old `msmart` package if necessary, and install this fork `msmart-ng`.
+
 ```shell
 pip uninstall msmart
 pip install msmart-ng
 ```
 
 ## Usage
-Discover all devices on the LAN with the `midea-discover` command.
+### CLI
+A simple command line interface is provided to discover and query devices. 
+
+```shell
+$ msmart-ng --help
+usage: msmart-ng [-h] [-v] {discover,query} ...
+
+Command line utility for msmart-ng.
+
+options:
+  -h, --help        show this help message and exit
+  -v, --version     show program's version number and exit
+
+Command:
+  {discover,query}
+```
+
+Each subcommand has additional help available. e.g. `msmart-ng discover --help`
+
+#### Discover
+Discover all devices on the LAN with the `msmart-ng discover` subcommand. 
 
 ```shell
-$ midea-discover 
-INFO:msmart.cli:msmart version: 2023.9.0
-INFO:msmart.cli:Only supports AC devices. Only supports MSmartHome and 美的美居.
+$ msmart-ng discover
+INFO:msmart.cli:Discovering all devices on local network.
 ...
 INFO:msmart.cli:Found 2 devices.
 INFO:msmart.cli:Found device:
 {'ip': '10.100.1.140', 'port': 6444, 'id': 15393162840672, 'online': True, 'supported': True, 'type': <DeviceType.AIR_CONDITIONER: 172>, 'name': 'net_ac_F7B4', 'sn': '000000P0000000Q1F0C9D153F7B40000', 'key': None, 'token': None}
 INFO:msmart.cli:Found device:
 {'ip': '10.100.1.239', 'port': 6444, 'id': 147334558165565, 'online': True, 'supported': True, 'type': <DeviceType.AIR_CONDITIONER: 172>, 'name': 'net_ac_63BA', 'sn': '000000P0000000Q1B88C29C963BA0000', 'key': '3a13f53f335042f9ae5fd266a6bd779459ed7ee7e09842f1a0e03c024890fc96', 'token': '56a72747cef14d55e17e69b46cd98deae80607e318a7b55cb86bb98974501034c657e39e4a4032e3c8cc9a3cab00fd3ec0bab4a816a57f68b8038977406b7431'}
 ```
+
 Check the output to ensure the type is 0xAC and the `supported` property is True.
 
 Save the device ID, IP address, and port. Version 3 devices will also require the `token` and `key` fields to control the device.
 
-#### Note: V1 Device Owners
+##### Note: V1 Device Owners
 Users with V1 devices will see the following error:
+
 ```
 ERROR:msmart.discover:V1 device not supported yet.
 ```
-I don't have any V1 devices to test with so please create an issue with the output of `midea-discover -d`.
 
-### Docker
-A docker image is available on ghcr.io at `ghcr.io/mill1000/msmart-ng`. The container should be run with `--network=host` to allow broadcast packets to reach devices on the local network. Additional arguments to the container are passed to the `midea-discover` command.
+I don't have any V1 devices to test with so please create an issue with the output of `msmart-ng discover --debug`.
 
-```shell
-$ docker run --network=host ghcr.io/mill1000/msmart-ng:latest --help
-usage: midea-discover [-h] [-d] [-a ACCOUNT] [-p PASSWORD] [-i IP] [-c COUNT] [--china]
+#### Query
+Query device state and capabilities with the `msmart-ng query` subcommand.
 
-Discover Midea devices and print device information.
+**Note:** Version 3 devices need to specify either the `--auto` argument or the `--token`, `--key` and `--id` arguments to make a connection.
+
+```shell
+$ msmart-ng query <HOST>
 
-options:
-  -h, --help            show this help message and exit
-  -d, --debug           Enable debug logging. (default: False)
-  -a ACCOUNT, --account ACCOUNT
-                        MSmartHome or 美的美居 account username. (default: [email protected])
-  -p PASSWORD, --password PASSWORD
-                        MSmartHome or 美的美居 account password. (default: lovemidea4ever)
-  -i IP, --ip IP        IP address of a device. Useful if broadcasts don't work, or to query a single device. (default: None)
-  -c COUNT, --count COUNT
-                        Number of broadcast packets to send. (default: 3)
-  --china               Use China server. (default: False)
 ```
 
+Device capabilities can be queried with the `--capabilities` argument.
+
 ### Home Assistant
 Use [this fork](https://github.com/mill1000/midea-ac-py) of midea-ac-py to control devices from Home Assistant.
 
 ### Python
 See the included [example](example.py) for controlling devices from a script.
 
+## Docker
+A docker image is available on ghcr.io at `ghcr.io/mill1000/msmart-ng`. The container should be run with `--network=host` to allow broadcast packets to reach devices on the local network. Additional arguments to the container are passed to the `msmart-ng` CLI.
+
+```shell
+$ docker run --network=host ghcr.io/mill1000/msmart-ng:latest --help
+usage: msmart-ng [-h] [-v] {discover,query} ...
+
+Command line utility for msmart-ng.
+
+options:
+  -h, --help        show this help message and exit
+  -v, --version     show program's version number and exit
+
+Command:
+  {discover,query}
+```
+
 ## Gratitude
 This project is a fork of [mac-zhou/midea-msmart](https://github.com/mac-zhou/midea-msmart), and builds upon the work of
 * [dudanov/MideaUART](https://github.com/dudanov/MideaUART)

diff --git a/msmart/base_device.py b/msmart/base_device.py
@@ -121,8 +121,8 @@ def online(self) -> bool:
     def supported(self) -> bool:
         return self._supported
 
-    def __str__(self) -> str:
-        return str({
+    def to_dict(self) -> dict:
+        return {
             "ip": self.ip,
             "port": self.port,
             "id": self.id,
@@ -133,4 +133,7 @@ def __str__(self) -> str:
             "sn": self.sn,
             "key": self.key,
             "token": self.token
-        })
+        }
+
+    def __str__(self) -> str:
+        return str(self.to_dict())
diff --git a/msmart/cli.py b/msmart/cli.py
@@ -1,60 +1,103 @@
 import argparse
 import asyncio
 import logging
+from typing import NoReturn
 
 from msmart import __version__
 from msmart.const import OPEN_MIDEA_APP_ACCOUNT, OPEN_MIDEA_APP_PASSWORD
+from msmart.device import AirConditioner as AC
+from msmart.device import Device
 from msmart.discover import Discover
 
 _LOGGER = logging.getLogger(__name__)
 
 
-async def _discover(ip: str, count: int, account: str, password: str, china: bool, **_kwargs) -> None:
+async def _discover(args) -> None:
     """Discover Midea devices and print configuration information."""
 
-    _LOGGER.info("msmart version: %s", __version__)
-    _LOGGER.info(
-        "Only supports AC devices. Only supports MSmartHome and 美的美居.")
-
-    if china and (account == OPEN_MIDEA_APP_ACCOUNT or password == OPEN_MIDEA_APP_PASSWORD):
-        _LOGGER.error(
-            "To use China server set account (phone number) and password of 美的美居.")
-        exit(1)
-
     devices = []
-    if ip is None or ip == "":
-        devices = await Discover.discover(account=account, password=password, discovery_packets=count)
+    if args.host is None:
+        _LOGGER.info("Discovering all devices on local network.")
+        devices = await Discover.discover(account=args.account, password=args.password, discovery_packets=args.count)
     else:
-        dev = await Discover.discover_single(ip, account=account, password=password, discovery_packets=count)
+        _LOGGER.info("Discovering %s on local network.", args.host)
+        dev = await Discover.discover_single(args.host, account=args.account, password=args.password, discovery_packets=args.count)
         if dev:
             devices.append(dev)
 
     if len(devices) == 0:
         _LOGGER.error("No devices found.")
         return
 
+    # Dump only basic device info from the base class
     _LOGGER.info("Found %d devices.", len(devices))
     for device in devices:
-        _LOGGER.info("Found device:\n%s", device)
 
+        if isinstance(device, AC):
+            device = super(AC, device)
+
+        _LOGGER.info("Found device:\n%s", device.to_dict())
+
+
+async def _query(args) -> None:
+    """Query device state or capabilities."""
+
+    if args.auto and (args.token or args.key or args.device_id):
+        _LOGGER.warning(
+            "--token, --key and --id are ignored with --auto option.")
+
+    if args.auto:
+        # Use discovery to automatically connect and authenticate with device
+        _LOGGER.info("Discovering %s on local network.", args.host)
+        device = await Discover.discover_single(args.host, account=args.account, password=args.password)
+
+        if device is None:
+            _LOGGER.error("Device not found.")
+            exit(1)
+    else:
+        # Manually create device and authenticate
+        device = AC(ip=args.host, port=6444, device_id=args.device_id)
+        if args.token and args.key:
+            await device.authenticate(args.token, args.key)
+
+    if not isinstance(device, AC):
+        _LOGGER.error("Device is not supported.")
+        exit(1)
+
+    if args.capabilities:
+        _LOGGER.info("Querying device capabilities.")
+        await device.get_capabilities()
+
+        if not device.online:
+            _LOGGER.error("Device is not online.")
+            exit(1)
+
+        # TODO method to get caps in string format
+        _LOGGER.info("%s", str({
+            "supported_modes": device.supported_operation_modes,
+            "supported_swing_modes": device.supported_swing_modes,
+            "supports_eco_mode": device.supports_eco_mode,
+            "supports_turbo_mode": device.supports_turbo_mode,
+            "supports_freeze_protection_mode": device.supports_freeze_protection_mode,
+            "supports_display_control": device.supports_display_control,
+            "max_target_temperature": device.max_target_temperature,
+            "min_target_temperature": device.min_target_temperature,
+        }))
+    else:
+        _LOGGER.info("Querying device state.")
+        await device.refresh()
+
+        if not device.online:
+            _LOGGER.error("Device is not online.")
+            exit(1)
+
+        _LOGGER.info("%s", device)
 
-def main() -> None:
-    parser = argparse.ArgumentParser(description="Discover Midea devices and print device information.",
-                                     formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument(
-        "-d", "--debug", help="Enable debug logging.", action="store_true")
-    parser.add_argument(
-        "-a", "--account", help="MSmartHome or 美的美居 account username.", default=OPEN_MIDEA_APP_ACCOUNT)
-    parser.add_argument(
-        "-p", "--password", help="MSmartHome or 美的美居 account password.", default=OPEN_MIDEA_APP_PASSWORD)
-    parser.add_argument(
-        "-i", "--ip", help="IP address of a device. Useful if broadcasts don't work, or to query a single device.")
-    parser.add_argument(
-        "-c", "--count", help="Number of broadcast packets to send.", default=3, type=int)
-    parser.add_argument("--china", help="Use China server.",
-                        action="store_true")
-    args = parser.parse_args()
 
+def _run(args) -> NoReturn:
+    """Helper method to setup logging, validate args and execute the desired function."""
+
+    # Configure logging
     if args.debug:
         logging.basicConfig(level=logging.DEBUG)
         # Keep httpx as info level
@@ -66,11 +109,120 @@ def main() -> None:
         logging.getLogger("httpx").setLevel(logging.WARNING)
         logging.getLogger("httpcore").setLevel(logging.WARNING)
 
+    # Validate common arguments
+    if args.china and (args.account == OPEN_MIDEA_APP_ACCOUNT or args.password == OPEN_MIDEA_APP_PASSWORD):
+        _LOGGER.error(
+            "Account (phone number) and password of 美的美居 is required to use --china option.")
+        exit(1)
+
     try:
-        asyncio.run(_discover(**vars(args)))
+        asyncio.run(args.func(args))
     except KeyboardInterrupt:
         pass
 
+    exit(0)
+
+
+def main() -> NoReturn:
+    """Main entry point for msmart-ng command."""
+
+    # Define the main parser to select subcommands
+    parser = argparse.ArgumentParser(
+        description="Command line utility for msmart-ng."
+    )
+    parser.add_argument("-v", "--version",
+                        action="version", version=f"msmart version: {__version__}")
+    subparsers = parser.add_subparsers(title="Command", dest="command",
+                                       required=True)
+
+    # Define some common arguments
+    common_parser = argparse.ArgumentParser(add_help=False)
+    common_parser.add_argument("-d", "--debug",
+                               help="Enable debug logging.", action="store_true")
+    common_parser.add_argument("--account",
+                               help="MSmartHome or 美的美居 username for discovery and automatic authentication",
+                               default=OPEN_MIDEA_APP_ACCOUNT)
+    common_parser.add_argument("--password",
+                               help="MSmartHome or 美的美居 password for discovery and automatic authentication.",
+                               default=OPEN_MIDEA_APP_PASSWORD)
+    common_parser.add_argument("--china",
+                               help="Use China server for discovery and automatic authentication.",
+                               action="store_true")
+
+    # Setup discover parser
+    discover_parser = subparsers.add_parser("discover",
+                                            description="Discover device(s) on the local network.",
+                                            parents=[common_parser])
+    discover_parser.add_argument("host",
+                                 help="Hostname or IP address of a single device to discover.",
+                                 nargs="?", default=None)
+    discover_parser.add_argument("--count",
+                                 help="Number of broadcast packets to send.",
+                                 default=3, type=int)
+    discover_parser.set_defaults(func=_discover)
+
+    # Setup query parser
+    query_parser = subparsers.add_parser("query",
+                                         description="Query information from a device on the local network.",
+                                         parents=[common_parser])
+    query_parser.add_argument("host",
+                              help="Hostname or IP address of device.")
+    query_parser.add_argument("--capabilities",
+                              help="Query device capabilities instead of state.",
+                              action="store_true")
+    query_parser.add_argument("--auto",
+                              help="Automatically authenticate V3 devices.",
+                              action="store_true")
+    query_parser.add_argument("--id",
+                              help="Device ID for V3 devices.",
+                              dest="device_id", type=int, default=0)
+    query_parser.add_argument("--token",
+                              help="Authentication token for V3 devices.",
+                              type=bytes.fromhex)
+    query_parser.add_argument("--key",
+                              help="Authentication key for V3 devices.",
+                              type=bytes.fromhex)
+    query_parser.set_defaults(func=_query)
+
+    # Run with args
+    _run(parser.parse_args())
+
+
+def _legacy_main() -> NoReturn:
+    """Main entry point for legacy midea-discover command."""
+
+    async def _wrap_discover(args) -> None:
+        """Wrapper method to mimic legacy behavior."""
+        # Map old args to new names as needed
+        args.host = args.ip
+
+        # Output legacy information
+        _LOGGER.info("msmart version: %s", __version__)
+        _LOGGER.info(
+            "Only supports AC devices. Only supports MSmartHome and 美的美居.")
+
+        await _discover(args)
+
+    parser = argparse.ArgumentParser(
+        description="Discover Midea devices and print device information.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument(
+        "-d", "--debug", help="Enable debug logging.", action="store_true")
+    parser.add_argument(
+        "-a", "--account", help="MSmartHome or 美的美居 account username.", default=OPEN_MIDEA_APP_ACCOUNT)
+    parser.add_argument(
+        "-p", "--password", help="MSmartHome or 美的美居 account password.", default=OPEN_MIDEA_APP_PASSWORD)
+    parser.add_argument(
+        "-i", "--ip", help="IP address of a device. Useful if broadcasts don't work, or to query a single device.")
+    parser.add_argument(
+        "-c", "--count", help="Number of broadcast packets to send.", default=3, type=int)
+    parser.add_argument("--china", help="Use China server.",
+                        action="store_true")
+    parser.set_defaults(func=_wrap_discover)
+
+    # Run with args
+    _run(parser.parse_args())
+
 
 if __name__ == "__main__":
     main()