dymo-bluetooth/dymo_bluetooth/bluetooth.py

# MIT License
# 
# Copyright (c) 2024 ysfchn / Yusuf Cihan
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import asyncio
from io import BytesIO
from pathlib import Path
from PIL import Image, ImageChops
from bleak import BleakScanner, BleakClient
from typing import List, TYPE_CHECKING
from dymo_bluetooth.printer import Canvas, Result, command_print

if TYPE_CHECKING:
    from bleak.backends.device import BLEDevice
    from bleak.backends.characteristic import BleakGATTCharacteristic


SERVICE_UUID = "be3dd650-{uuid}-42f1-99c1-f0f749dd0678".format(uuid = "2b3d")
PRINT_REQUEST_UUID = "be3dd651-{uuid}-42f1-99c1-f0f749dd0678".format(uuid = "2b3d")
PRINT_REPLY_UUID = "be3dd652-{uuid}-42f1-99c1-f0f749dd0678".format(uuid = "2b3d")

# Not used in the actual vendor app.
UNKNOWN_UUID = "be3dd653-{uuid}-42f1-99c1-f0f749dd0678".format(uuid = "2b3d")


def is_espressif(mac : str):
    """
    Returns True if given MAC address is in the range of Espressif Inc.
    """
    start_block = 0x58_CF_79 << 24
    end_block = start_block + (1 << 24) # Exclusive
    mac_value = int(mac.replace(":", ""), base = 16)
    if (mac_value >= start_block) and (mac_value < end_block):
        return True
    return False


class Printer:
    def __init__(self, impl : "BLEDevice") -> None:
        self._impl = impl
        self._client = BleakClient(self._impl)
    
    async def connect(self):
        if self._client.is_connected:
            return
        await self._client.connect()
    
    async def disconnect(self):
        if not self._client.is_connected:
            return
        await self._client.disconnect()

    async def print(self, canvas : Canvas):
        if not self._client.is_connected:
            raise Exception("Printer is not connected!")
        print_request : "BleakGATTCharacteristic" = self._client.services.get_characteristic(PRINT_REQUEST_UUID) # type: ignore # noqa: E501
        print_reply : "BleakGATTCharacteristic" = self._client.services.get_characteristic(PRINT_REPLY_UUID) # type: ignore # noqa: E501
        future : asyncio.Future[Result] = asyncio.Future()
        should_discard : bool = False
        # Printer sends two messages, first is the PRINTING, and second one is the 
        # printing result. So we discard the first message.
        async def reply_get(_, data : bytearray): # noqa: E501
            nonlocal should_discard
            result = Result.from_bytes(data)
            if (not should_discard) and (result.value in [0, 1]):
                should_discard = True
                return
            # This is the second reply, which holds the actual status code. 
            future.set_result(result)
        await self._client.start_notify(print_reply, reply_get)
        for chunk in command_print(canvas):
            await self._client.write_gatt_char(print_request, chunk, True)
        return await future


async def discover_printers(max_timeout : int = 5) -> List[Printer]:
    """
    Searches for printers nearby and returns a list of Printer objects. If no printer
    has found in the initial search, waits for scanning until the max timeout has been 
    reached.
    """
    printers : List[Printer] = []
    waited_total = 0
    async with BleakScanner(service_uuids = [SERVICE_UUID]) as scanner:
        while True:
            # TODO: In some cases, advetisement data may be non-null, containing
            # additional metadata about printer state but it is not implemented yet.
            for device, _ in scanner.discovered_devices_and_advertisement_data.values(): # noqa: E501
                if not is_espressif(device.address):
                    continue
                # Also match the "Letratag" name just to be sure that we are looking for
                # the right device, since there are also other products made by DYMO.
                if (device.name or "") == f"Letratag {device.address.replace(':', '')}":
                    printers.append(Printer(device))
            # Do we have any candidate printers? If so, return the found printers. 
            # Otherwise, wait for the next scans until we found any.
            if printers:
                break
            elif waited_total >= max_timeout:
                return []
            await asyncio.sleep(0.5)
            waited_total += 0.5
    return printers


def convert_image_to_canvas(
    image : Image.Image, 
    dither : bool = True,
    trim : bool = False
):
    """
    Converts an Pillow Image to a Canvas object.
    """
    output = image.convert("1", dither = \
        Image.Dither.FLOYDSTEINBERG if dither else Image.Dither.NONE
    )
    # If trim is enabled, discard trailing and leading blank lines.
    if trim:
        mask = Image.new("1", output.size, color = 255)
        diff = ImageChops.difference(output, mask)
        output = output.crop(diff.getbbox(alpha_only = False))
    # Shrink the image from the center if it exceeds the print height,
    # or max printable width.
    if (output.height > Canvas.HEIGHT):
        start_y = int(output.height / 2) - int(Canvas.HEIGHT / 2)
        output = output.crop(
            (0, start_y, output.width, start_y + Canvas.HEIGHT)
        )
    elif (output.height < Canvas.HEIGHT):
        raise ValueError("Image is too small, resizing not implemented.")
    if (output.width) > Canvas.MAX_WIDTH:
        raise ValueError("Image is too large, resizing not implemented.")
    # Convert image to pixel array.
    canvas = Canvas()
    for w in range(output.width):
        for h in range(output.height):
            pixel = output.getpixel((w, h, ))
            canvas.set_pixel(w, h, not pixel)
    return canvas


def create_image(path : Path, dither : bool = True):
    """
    Converts an image file in given path to Canvas.
    """
    buffer = BytesIO()
    with path.open("rb") as op:
        buffer.write(op.read())
    buffer.seek(0)
    image = Image.open(buffer)
    return convert_image_to_canvas(image, dither)


def create_code_128(text : str):
    """
    Creates a Code 128 barcode and dumps to Canvas.
    """
    try:
        from barcode import Code128 # type: ignore
        from barcode.writer import ImageWriter # type: ignore
    except ModuleNotFoundError:
        raise Exception("This method requires 'python-barcode' to be installed.")
    imwrite = ImageWriter()
    imwrite.dpi = 200
    code = Code128(text, writer = imwrite)
    return convert_image_to_canvas(code.render(text = ""), dither = False, trim = True)