Skip to content

Commit

Permalink
feat(cli): midealocal CLI tool (#204)
Browse files Browse the repository at this point in the history
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
  - Introduced a CLI tool for interacting with Midea local devices.
- Added device discovery, message loading, and protocol downloading
functionalities to the CLI tool.
- Configured a Python debugger for easier module discovery and
debugging.
  - Added constants for application account and password configuration.
  - Added a console script entry point for the CLI tool.

- **Chores**
  - Added `midea-local.json` to `.gitignore`.
  - Updated `requirements.txt` with the `platformdirs` package.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
  • Loading branch information
rokam authored Jul 9, 2024
1 parent 4da6e3e commit 236e33a
Show file tree
Hide file tree
Showing 5 changed files with 332 additions and 0 deletions.
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -134,3 +134,4 @@ reports/

# Optional config file for library_test.py script
library_test.json
midea-local.json
70 changes: 70 additions & 0 deletions .vscode/launch.json
Original file line number Diff line number Diff line change
@@ -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."
}
]
}
255 changes: 255 additions & 0 deletions midealocal/cli.py
Original file line number Diff line number Diff line change
@@ -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()
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,4 @@ build
defusedxml
ifaddr
pycryptodome
platformdirs
5 changes: 5 additions & 0 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down

0 comments on commit 236e33a

Please sign in to comment.