Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(cli): midealocal CLI tool #204

Merged
merged 9 commits into from
Jul 9, 2024
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .vscode/launch.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Discover",
"type": "debugpy",
"request": "launch",
"module": "midealocal.cli",
"args": ["discover"]
}
]
}
219 changes: 219 additions & 0 deletions midealocal/cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,219 @@
"""Midea local CLI."""

import asyncio
import contextlib
import inspect
import logging
import sys
from argparse import ArgumentParser, Namespace
from typing import Any, NoReturn

import aiohttp
from colorlog import ColoredFormatter

from midealocal.cloud import clouds, get_midea_cloud
from midealocal.const import OPEN_MIDEA_APP_ACCOUNT, OPEN_MIDEA_APP_PASSWORD
from midealocal.device import ProtocolVersion
from midealocal.devices import device_selector
from midealocal.discover import discover
from midealocal.version import __version__

_LOGGER = logging.getLogger("cli")


async def _get_keys(args: Namespace, device_id: int) -> dict[int, dict[str, Any]]:
session = aiohttp.ClientSession()
with 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="",
)

if dev.connect():
_LOGGER.info("Found device:\n%s", dev.attributes)
break
rokam marked this conversation as resolved.
Show resolved Hide resolved


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",
rokam marked this conversation as resolved.
Show resolved Hide resolved
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 _download(args: Namespace) -> None:
"""Download a device's protocol implementation from the cloud."""
# Use discovery to to find device information
_LOGGER.info("Discovering %s on local network.", args.host)
rokam marked this conversation as resolved.
Show resolved Hide resolved


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",
default=OPEN_MIDEA_APP_ACCOUNT,
)
common_parser.add_argument(
"--password",
"-p",
type=str,
help="Set cloud password",
default=OPEN_MIDEA_APP_PASSWORD,
)
common_parser.add_argument(
"--cloud-name",
"-cn",
type=str,
help="Set Cloud name",
choices=clouds.keys(),
default="MSmartHome",
rokam marked this conversation as resolved.
Show resolved Hide resolved
)

# 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.",
required=False,
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)

# Run with args
_run(parser.parse_args())


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()
3 changes: 3 additions & 0 deletions midealocal/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,6 @@

MAX_BYTE_VALUE = 0xFF
MAX_DOUBLE_BYTE_VALUE = 0xFFFF

OPEN_MIDEA_APP_ACCOUNT = "[email protected]"
OPEN_MIDEA_APP_PASSWORD = "this_is_a_password1" # noqa: S105
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