commands.py

cli.commands

CLI Interface.

Attributes

all_attrs module-attribute

all_attrs = list(ember_mug.consts.ATTR_LABELS) + list(
    ember_mug.consts.EXTRA_ATTRS
)

get_attribute_names module-attribute

get_attribute_names = [
    (n.replace("_", "-")) for n in cli.commands.all_attrs
]

Classes

EmberMugCli

Very simple CLI Interface to interact with a mug.

Source code in ember_mug/cli/commands.py

class EmberMugCli:
    """Very simple CLI Interface to interact with a mug."""

    _commands: ClassVar[dict[str, Callable[[Namespace], Awaitable]]] = {
        "find": find_device_cmd,
        "discover": discover_cmd,
        "info": fetch_info_cmd,
        "poll": poll_device_cmd,
        "get": get_device_value_cmd,
        "set": set_device_value_cmd,
    }

    def __init__(self) -> None:
        """Create parsers."""
        self.parser = ArgumentParser(prog="ember-mug", description="CLI to interact with an Ember Mug")
        shared_parser = ArgumentParser(add_help=False)
        shared_parser.add_argument(
            "-m",
            "--mac",
            action="store",
            type=validate_mac,
            help="Only look for this specific address",
        )
        self.parser.add_argument(
            "-d",
            "--debug",
            action="store_true",
            help="Print extra information for development or debugging issues",
        )
        self.parser.add_argument(
            "--log-file",
            type=Path,
            nargs="?",
            help="File to write logs too (Will be overwritten)",
        )
        self.parser.add_argument("--version", action="version", version=__version__)
        shared_parser.add_argument("-r", "--raw", help="No formatting. One value per line.", action="store_true")
        if IS_LINUX is True:
            # Only works on Linux with BlueZ so don't add for others.
            shared_parser.add_argument(
                "-a",
                "--adapter",
                action="store",
                help="Use this Bluetooth adapter instead of the default one (for Bluez)",
            )
        subparsers = self.parser.add_subparsers(dest="command", required=True)
        subparsers.add_parser("find", description="Find the first paired device", parents=[shared_parser])
        subparsers.add_parser("discover", description="Discover devices in pairing mode", parents=[shared_parser])
        info_parsers = ArgumentParser(add_help=False)
        info_parsers.add_argument("-e", "--extra", help="Show extra info", action="store_true")
        info_parsers.add_argument("--imperial", help="Use Imperial units", action="store_true")
        subparsers.add_parser("info", description="Fetch all info from device", parents=[shared_parser, info_parsers])
        subparsers.add_parser("poll", description="Poll device for information", parents=[shared_parser, info_parsers])
        get_parser = subparsers.add_parser("get", description="Get mug value", parents=[shared_parser, info_parsers])
        get_parser.add_argument(dest="attributes", metavar="ATTRIBUTE", choices=get_attribute_names, nargs="+")
        set_parser = subparsers.add_parser("set", description="Set mug value", parents=[shared_parser, info_parsers])
        set_parser.add_argument("--name", help="Name", required=False)
        set_parser.add_argument("--target-temp", help="Target Temperature", type=float, required=False)
        set_parser.add_argument("--temperature-unit", help="Temperature Unit", choices=["C", "F"], required=False)
        set_parser.add_argument("--led-colour", help="LED Colour", type=colour_type, required=False)
        set_parser.add_argument(
            "--volume-level",
            help="Volume Level",
            choices=[v.value for v in VolumeLevel],
            required=False,
        )

    async def run(self) -> None:
        """Run the specified command based on subparser."""
        args = self.parser.parse_args()
        out_file: IO[str] | None = None
        if IS_LINUX is False:
            args.adapter = None  # Set for other platforms
        if args.debug:
            if args.log_file:
                try:
                    out_file = args.log_file.open("w")
                except OSError as e:
                    self.parser.error(f"Failed to open '{args.log_file}': {e}")

            logging.basicConfig(
                stream=out_file or sys.stdout,
                level=logging.DEBUG,
                format="[%(asctime)s] %(levelname)s [%(filename)s.%(funcName)s:%(lineno)d] %(message)s",
            )
        try:
            await self._commands[args.command](args)
        finally:
            if out_file is not None:
                out_file.close()

Attributes

parser instance-attribute

parser = argparse.ArgumentParser(
    prog="ember-mug",
    description="CLI to interact with an Ember Mug",
)

Functions

run async

run() -> None

Run the specified command based on subparser.

Source code in ember_mug/cli/commands.py

async def run(self) -> None:
    """Run the specified command based on subparser."""
    args = self.parser.parse_args()
    out_file: IO[str] | None = None
    if IS_LINUX is False:
        args.adapter = None  # Set for other platforms
    if args.debug:
        if args.log_file:
            try:
                out_file = args.log_file.open("w")
            except OSError as e:
                self.parser.error(f"Failed to open '{args.log_file}': {e}")

        logging.basicConfig(
            stream=out_file or sys.stdout,
            level=logging.DEBUG,
            format="[%(asctime)s] %(levelname)s [%(filename)s.%(funcName)s:%(lineno)d] %(message)s",
        )
    try:
        await self._commands[args.command](args)
    finally:
        if out_file is not None:
            out_file.close()

Functions

colour_type

colour_type(value: str) -> ember_mug.data.Colour

Convert a hex or rgb colour to a Colour object.

Source code in ember_mug/cli/commands.py

def colour_type(value: str) -> Colour:
    """Convert a hex or rgb colour to a Colour object."""
    print(value)
    if match := re.match(r"#?([0-9a-f]{6}([0-9a-f]{2})?)", value, re.IGNORECASE):
        raw_colours = match.group(1)
        colours = [
            255 if (colour := raw_colours[i : i + 2]) is None else int(colour, 16)
            for i in range(0, len(raw_colours), 2)
        ]
        return Colour(*colours)

    with contextlib.suppress(ValueError, AssertionError):
        colours = [int(v) for v in value.split(",")]
        if len(colours) not in (3, 4):
            raise ArgumentTypeError("Three or four values should be specified for colour")
        if not all(0 <= c <= 255 for c in colours):
            raise ArgumentTypeError("Colour values must be between 0 and 255")
        return Colour(*colours)

    msg = f'"{value}" is not a valid rgba or hex colour'
    raise ArgumentTypeError(msg)

discover_cmd async

discover_cmd(
    args: argparse.Namespace,
) -> list[
    tuple[
        bleak.backends.device.BLEDevice,
        bleak.AdvertisementData,
    ]
]

Discover new devices in pairing mode.

Source code in ember_mug/cli/commands.py

async def discover_cmd(args: Namespace) -> list[tuple[BLEDevice, AdvertisementData]]:
    """Discover new devices in pairing mode."""
    try:
        mugs = await discover_devices(mac=args.mac)
    except BleakError as e:
        print(f"An error occurred trying to discover devices: {e}")
        sys.exit(1)
    if not mugs:
        print('No devices were found. Be sure it is in pairing mode. Or use "find" if already paired.')
        sys.exit(1)

    for mug, advertisement in mugs:
        if args.raw:
            print(mug.address)
        else:
            model_info = get_model_info_from_advertiser_data(advertisement)
            model_number = model_info.model.value if model_info.model else "Unknown Model"
            print(f"Found {model_info.device_type.value}:", mug)
            print("Name:", advertisement.local_name)
            print("Model:", f"{model_info.name} [{model_number}]")
            print("Colour:", model_info.colour.value if model_info.colour else "Unknown")
            print("Capacity:", format_capacity(model_info.capacity))
    return mugs

fetch_info_cmd async

fetch_info_cmd(args: argparse.Namespace) -> None

Fetch all information from a mug and end.

Source code in ember_mug/cli/commands.py

async def fetch_info_cmd(args: Namespace) -> None:
    """Fetch all information from a mug and end."""
    mug = await get_device(args)
    async with mug.connection(adapter=args.adapter):
        if not args.raw:
            print("Connected.\nFetching Info")
        await mug.update_all()
    print_info(mug)

find_device_cmd async

find_device_cmd(
    args: argparse.Namespace,
) -> tuple[
    bleak.backends.device.BLEDevice, bleak.AdvertisementData
]

Find a single device that has already been paired.

Source code in ember_mug/cli/commands.py

async def find_device_cmd(args: Namespace) -> tuple[BLEDevice, AdvertisementData]:
    """Find a single device that has already been paired."""
    try:
        device, advertisement = await find_device(mac=args.mac, adapter=args.adapter)
    except BleakError as e:
        print(f"An error occurred trying to find a device: {e}")
        sys.exit(1)
    if not device or not advertisement:
        print("No device was found.")
        sys.exit(1)
    if not args.raw:
        print("Found device:", device)
    return device, advertisement

get_device async

get_device(
    args: argparse.Namespace,
) -> ember_mug.mug.EmberMug

Help to get the devices based on command args.

Source code in ember_mug/cli/commands.py

async def get_device(args: Namespace) -> EmberMug:
    """Help to get the devices based on command args."""
    device, advertisement = await find_device_cmd(args)
    manufacturer_data = advertisement.manufacturer_data
    model_info = get_model_info_from_advertiser_data(advertisement)
    if model_info.model == DeviceModel.UNKNOWN_DEVICE and not args.raw:
        data = manufacturer_data.get(EMBER_BLE_SIG, manufacturer_data.get(TESTING_BLE_SIG, None))
        print(f"Warning: No model found matching advertisement data: {data!r}")

    mug = EmberMug(
        device,
        model_info,
        use_metric=not args.imperial,
        debug=args.debug,
    )
    if not args.raw:
        print("Connecting...")
    return mug

get_device_value_cmd async

get_device_value_cmd(args: argparse.Namespace) -> None

Get values from the mug and print them.

Source code in ember_mug/cli/commands.py

async def get_device_value_cmd(args: Namespace) -> None:
    """Get values from the mug and print them."""
    mug = await get_device(args)
    data = {}
    attributes = [a.replace("-", "_") for a in args.attributes]
    async with mug.connection(adapter=args.adapter):
        for attr in attributes:
            try:
                value = await getattr(mug, f"get_{attr}")()
            except NotImplementedError as e:
                print(e)
                sys.exit(1)
            setattr(mug.data, attr, value)
            data[attr] = value
    if args.raw:
        print("\n".join(str(v) for v in data.values()))
    else:
        print_table([(ATTR_LABELS.get(attr, attr), str(mug.data.get_formatted_attr(attr))) for attr in data])

poll_device_cmd async

poll_device_cmd(args: argparse.Namespace) -> None

Fetch all information and keep polling for changes.

Source code in ember_mug/cli/commands.py

async def poll_device_cmd(args: Namespace) -> None:
    """Fetch all information and keep polling for changes."""
    mug = await get_device(args)
    async with mug.connection(adapter=args.adapter):
        if not args.raw:
            print("Connected.\nFetching Info")
        await mug.update_all()
        print_info(mug)
        if not args.raw:
            print("\nWatching for changes")
        for _ in CommandLoop():
            for _ in range(60):
                await asyncio.sleep(1)
                print_changes(await mug.update_queued_attributes(), mug.data.user_unit)
            # Every minute do a full update
            print_changes(await mug.update_all(), mug.data.user_unit)

set_device_value_cmd async

set_device_value_cmd(args: argparse.Namespace) -> None

Set one or more values on the device.

Source code in ember_mug/cli/commands.py

async def set_device_value_cmd(args: Namespace) -> None:
    """Set one or more values on the device."""
    attrs = ("name", "target_temp", "temperature_unit", "led_colour", "volume_level")
    values = [(attr, value) for attr in attrs if (value := getattr(args, attr, None))]
    if not values:
        print("Please specify at least one attribute and value to set.")
        options = [f"--{a.replace('_', '-')}" for a in attrs]
        print(f"Options: {', '.join(options)}")
        sys.exit(1)

    mug = await get_device(args)
    async with mug.connection(adapter=args.adapter):
        for attr, value in values:
            method = getattr(mug, f"set_{attr.replace('-', '_')}")
            print(f"Setting {attr} to {value}")
            try:
                await method(value)
            except NotImplementedError as e:
                print(e)
                sys.exit(1)