diff --git a/.gitignore b/.gitignore index 2334a386..6043f058 100644 --- a/.gitignore +++ b/.gitignore @@ -134,3 +134,4 @@ reports/ # Optional config file for library_test.py script library_test.json +midea-local.json diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 00000000..779f725b --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,70 @@ +{ + "version": "0.2.0", + "configurations": [ + { + "name": "Python Debugger: Discover", + "type": "debugpy", + "request": "launch", + "module": "midealocal.cli", + "args": ["discover", "-d", "--host", "${input:host}"] + }, + { + "name": "Python Debugger: Decode message", + "type": "debugpy", + "request": "launch", + "module": "midealocal.cli", + "args": ["decode", "-d", "${input:message}"] + }, + { + "name": "Python Debugger: Save config", + "type": "debugpy", + "request": "launch", + "module": "midealocal.cli", + "args": [ + "save", + "--cloud-name", + "${input:cloud-name}", + "--username", + "${input:username}", + "--password", + "${input:password}" + ] + } + ], + "inputs": [ + { + "id": "host", + "type": "promptString", + "description": "Enter the host IP address (leave empty to broadcast on the network)", + "default": "" + }, + { + "id": "message", + "type": "promptString", + "description": "Enter hex message, including header and body." + }, + { + "id": "cloud-name", + "type": "pickString", + "options": [ + "美的美居", + "MSmartHome", + "Midea Air", + "NetHome Plus", + "Ariston Clima" + ], + "description": "Choose one of the cloud servers." + }, + { + "id": "username", + "type": "promptString", + "description": "Enter cloud username." + }, + { + "id": "password", + "type": "promptString", + "password": true, + "description": "Enter cloud password." + } + ] +} diff --git a/midealocal/cli.py b/midealocal/cli.py new file mode 100644 index 00000000..3176441b --- /dev/null +++ b/midealocal/cli.py @@ -0,0 +1,255 @@ +"""Midea local CLI.""" + +import asyncio +import contextlib +import inspect +import json +import logging +import sys +from argparse import ArgumentParser, Namespace +from pathlib import Path +from typing import Any, NoReturn + +import aiohttp +import platformdirs +from colorlog import ColoredFormatter + +from midealocal.cloud import clouds, get_midea_cloud +from midealocal.device import ProtocolVersion +from midealocal.devices import device_selector +from midealocal.discover import discover +from midealocal.exceptions import ElementMissing +from midealocal.version import __version__ + +_LOGGER = logging.getLogger("cli") + + +def get_config_file_path(relative: bool = False) -> Path: + """Get the config file path.""" + local_path = Path("midea-local.json") + if relative or local_path.exists(): + return local_path + return platformdirs.user_config_path(appname="midea-local").joinpath( + "midea-local.json", + ) + + +async def _get_keys(args: Namespace, device_id: int) -> dict[int, dict[str, Any]]: + if not args.cloud_name or not args.username or not args.password: + raise ElementMissing("Missing required parameters for cloud request.") + async with aiohttp.ClientSession() as session: + cloud = get_midea_cloud( + cloud_name=args.cloud_name, + session=session, + account=args.username, + password=args.password, + ) + + return await cloud.get_keys(device_id) + + +async def _discover(args: Namespace) -> None: + """Discover device information.""" + devices = discover(ip_address=args.host) + + 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.values(): + keys = ( + {0: {"token": "", "key": ""}} + if device["protocol"] != ProtocolVersion.V3 + else await _get_keys(args, device["device_id"]) + ) + + for key in keys.values(): + dev = device_selector( + name=device["device_id"], + device_id=device["device_id"], + device_type=device["type"], + ip_address=device["ip_address"], + port=device["port"], + token=key["token"], + key=key["key"], + protocol=device["protocol"], + model=device["model"], + subtype=0, + customize="", + ) + _LOGGER.debug("Trying to connect with key: %s", key) + if dev.connect(): + _LOGGER.info("Found device:\n%s", dev.attributes) + break + + _LOGGER.debug("Unable to connect with key: %s", key) + + +def _message(args: Namespace) -> None: + """Load message into device.""" + device_type = int(args.message[2]) + + device = device_selector( + device_id=0, + name="", + device_type=device_type, + ip_address="192.168.192.168", + port=6664, + protocol=ProtocolVersion.V2, + model="0000", + token="", + key="", + subtype=0, + customize="", + ) + + result = device.process_message(args.message) + + _LOGGER.info("Parsed message: %s", result) + + +def _save(args: Namespace) -> None: + data = { + "username": args.username, + "password": args.password, + "cloud_name": args.cloud_name, + } + json_data = json.dumps(data) + file = get_config_file_path(not args.user) + with file.open(mode="w+", encoding="utf-8") as f: + f.write(json_data) + + +def main() -> NoReturn: + """Launch main entry.""" + # Define the main parser to select subcommands + parser = ArgumentParser(description="Command line utility for midea-local.") + parser.add_argument( + "-v", + "--version", + action="version", + version=f"midea-local version: {__version__}", + ) + subparsers = parser.add_subparsers(title="Command", dest="command", required=True) + + # Define some common arguments + common_parser = ArgumentParser(add_help=False) + common_parser.add_argument( + "-d", + "--debug", + help="Enable debug logging.", + action="store_true", + ) + common_parser.add_argument( + "--username", + "-u", + type=str, + help="Set cloud username", + ) + common_parser.add_argument( + "--password", + "-p", + type=str, + help="Set cloud password", + ) + common_parser.add_argument( + "--cloud-name", + "-cn", + type=str, + help="Set Cloud name", + choices=clouds.keys(), + ) + + # 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.", + default=None, + ) + discover_parser.set_defaults(func=_discover) + + decode_msg_parser = subparsers.add_parser( + "decode", + description="Decode a message received to a device.", + parents=[common_parser], + ) + decode_msg_parser.add_argument( + "message", + help="Received message", + type=bytes.fromhex, + ) + decode_msg_parser.set_defaults(func=_message) + + save_parser = subparsers.add_parser( + "save", + description="Save config file with cloud parameters.", + parents=[common_parser], + ) + save_parser.add_argument( + "--user", + help="Save config file in your user config folder.", + action="store_true", + ) + save_parser.set_defaults(func=_save) + + config = get_config_file_path() + namespace = Namespace() + if config.exists(): + with config.open("r", encoding="utf-8") as f: + namespace = Namespace(**json.load(f)) + + # Run with args + _run(parser.parse_args(namespace=namespace)) + + +def _run(args: Namespace) -> NoReturn: + """Do setup logging, validate args and execute the desired function.""" + # Configure logging + if args.debug: + logging.basicConfig(level=logging.DEBUG) + # Keep httpx as info level + logging.getLogger("asyncio").setLevel(logging.INFO) + logging.getLogger("charset_normalizer").setLevel(logging.INFO) + else: + logging.basicConfig(level=logging.INFO) + # Set httpx to warning level + logging.getLogger("asyncio").setLevel(logging.WARNING) + logging.getLogger("charset_normalizer").setLevel(logging.WARNING) + + fmt = ( + "%(asctime)s.%(msecs)03d %(levelname)s (%(threadName)s) [%(name)s] %(message)s" + ) + colorfmt = f"%(log_color)s{fmt}%(reset)s" + logging.getLogger().handlers[0].setFormatter( + ColoredFormatter( + colorfmt, + datefmt="%Y-%m-%d %H:%M:%S", + reset=True, + log_colors={ + "DEBUG": "cyan", + "INFO": "green", + "WARNING": "yellow", + "ERROR": "red", + "CRITICAL": "red", + }, + ), + ) + + with contextlib.suppress(KeyboardInterrupt): + if inspect.iscoroutinefunction(args.func): + asyncio.run(args.func(args)) + else: + args.func(args) + + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt index 99bda5e6..1e3b1cc9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,4 @@ build defusedxml ifaddr pycryptodome +platformdirs diff --git a/setup.py b/setup.py index dbef07c0..1a6e8072 100644 --- a/setup.py +++ b/setup.py @@ -33,6 +33,11 @@ include=["midealocal", "midealocal.*"], exclude=["tests", "tests.*"], ), + entry_points={ + "console_scripts": [ + "midealocal = midealocal.cli:main", + ], + }, python_requires=">=3.11", classifiers=[ "Programming Language :: Python :: 3",