hardware.py

#!/usr/bin/python3
# -*-coding:utf-8-*-

import struct, zlib, logging
from bluetooth import BluetoothSocket, find_service, RFCOMM, discover_devices
from const import BtCommandByte
import codecs


class Paperang:
    standardKey = 0x35769521
    padding_line = 300
    max_send_msg_length = 2016
    max_recv_msg_length = 1024
    service_uuid = "00001101-0000-1000-8000-00805F9B34FB"

    def __init__(self, address=None):
        self.address = address
        self.crckeyset = False
        self.connected = True if self.connect() else False

    def connect(self):
        if self.address is None and not self.scandevices():
            return False
        if not self.scanservices():
            return False
        logging.info("Service found. Connecting to \"%s\" on %s..." % (self.service["name"], self.service["host"]))
        self.sock = BluetoothSocket(RFCOMM)
        self.sock.connect((self.service["host"], self.service["port"]))
        self.sock.settimeout(60)
        logging.info("Connected.")
        self.registerCrcKeyToBt()
        return True

    def disconnect(self):
        try:
            self.sock.close()
        except:
            pass
        logging.info("Disconnected.")

    def scandevices(self):
        logging.warning("Searching for devices...\n"
                        "It may take time, you'd better specify mac address to avoid a scan.")
        valid_names = ['MiaoMiaoJi', 'Paperang']
        nearby_devices = discover_devices(lookup_names=True)
        valid_devices = list(filter(lambda d: len(d) == 2 and d[1] in valid_names, nearby_devices))
        if len(valid_devices) == 0:
            logging.error("Cannot find device with name %s." % " or ".join(valid_names))
            return False
        elif len(valid_devices) > 1:
            logging.warning("Found multiple valid machines, the first one will be used.\n")
            logging.warning("\n".join(valid_devices))
        else:
            logging.warning(
                "Found a valid machine with MAC %s and name %s" % (valid_devices[0][0], valid_devices[0][1])
            )
        self.address = valid_devices[0][0]
        return True

    def scanservices(self):
        logging.info("Searching for services...")
        service_matches = find_service(uuid=self.service_uuid, address=self.address)
        valid_service = list(filter(
            lambda s: 'protocol' in s and 'name' in s and s['protocol'] == 'RFCOMM' and s['name'] == 'SerialPort',
            service_matches
        ))
        if len(valid_service) == 0:
            logging.error("Cannot find valid services on device with MAC %s." % self.address)
            return False
        logging.info("Found a valid service")
        self.service = valid_service[0]
        return True

    def sendMsgAllPackage(self, msg):
        # Write data directly to device
        sent_len = self.sock.send(msg)
        logging.info("Sending msg with length = %d..." % sent_len)

    def crc32(self, content):
        return zlib.crc32(content, self.crcKey if self.crckeyset else self.standardKey) & 0xffffffff

    def packPerBytes(self, bytes, control_command, i):
        result = struct.pack('<BBB', 2, control_command, i)
        result += struct.pack('<H', len(bytes))
        result += bytes
        result += struct.pack('<I', self.crc32(bytes))
        result += struct.pack('<B', 3)
        return result

    def addBytesToList(self, bytes):
        length = self.max_send_msg_length
        result = [bytes[i:i + length] for i in range(0, len(bytes), length)]
        return result

    def sendToBt(self, data_bytes, control_command, need_reply=True):
        bytes_list = self.addBytesToList(data_bytes)
        for i, bytes in enumerate(bytes_list):
            tmp = self.packPerBytes(bytes, control_command, i)
            self.sendMsgAllPackage(tmp)
        if need_reply:
            return self.recv()

    def recv(self):
        # Here we assume that there is only one received packet.
        raw_msg = self.sock.recv(self.max_recv_msg_length)
        parsed = self.resultParser(raw_msg)
        logging.info("Recv: " + codecs.encode(raw_msg, "hex_codec").decode())
        logging.info("Received %d packets: " % len(parsed) + "".join([str(p) for p in parsed]))
        return raw_msg, parsed

    def resultParser(self, data):
        base = 0
        res = []
        while base < len(data) and data[base] == '\x02':
            class Info(object):
                def __str__(self):
                    return "\nControl command: %s(%s)\nPayload length: %d\nPayload(hex): %s" % (
                        self.command, BtCommandByte.findCommand(self.command)
                        , self.payload_length, codecs.encode(self.payload, "hex_codec")
                    )

            info = Info()
            _, info.command, _, info.payload_length = struct.unpack('<BBBH', data[base:base + 5])
            info.payload = data[base + 5: base + 5 + info.payload_length]
            info.crc32 = data[base + 5 + info.payload_length: base + 9 + info.payload_length]
            base += 10 + info.payload_length
            res.append(info)
        return res

    def registerCrcKeyToBt(self, key=0x6968634 ^ 0x2e696d):
        logging.info("Setting CRC32 key...")
        msg = struct.pack('<I', int(key ^ self.standardKey))
        self.sendToBt(msg, BtCommandByte.PRT_SET_CRC_KEY)
        self.crcKey = key
        self.crckeyset = True
        logging.info("CRC32 key set.")

    def sendPaperTypeToBt(self, paperType=0):
        # My guess:
        # paperType=0: normal paper
        # paperType=1: official paper
        msg = struct.pack('<B', paperType)
        self.sendToBt(msg, BtCommandByte.PRT_SET_PAPER_TYPE)

    def sendPowerOffTimeToBt(self, poweroff_time=0):
        msg = struct.pack('<H', poweroff_time)
        self.sendToBt(msg, BtCommandByte.PRT_SET_POWER_DOWN_TIME)

    def sendImageToBt(self, binary_img):
        self.sendPaperTypeToBt()
        # msg = struct.pack("<%dc" % len(binary_img), *map(bytes, binary_img))
        msg = b"".join(map(lambda x: struct.pack("<c",x.to_bytes(1,byteorder="little")),binary_img))
        self.sendToBt(msg, BtCommandByte.PRT_PRINT_DATA, need_reply=False)
        self.sendFeedLineToBt(self.padding_line)

    def sendSelfTestToBt(self):
        msg = struct.pack('<B', 0)
        self.sendToBt(msg, BtCommandByte.PRT_PRINT_TEST_PAGE)

    def sendDensityToBt(self, density):
        msg = struct.pack('<B', density)
        self.sendToBt(msg, BtCommandByte.PRT_SET_HEAT_DENSITY)

    def sendFeedLineToBt(self, length):
        msg = struct.pack('<H', length)
        self.sendToBt(msg, BtCommandByte.PRT_FEED_LINE)

    def queryBatteryStatus(self):
        msg = struct.pack('<B', 1)
        self.sendToBt(msg, BtCommandByte.PRT_GET_BAT_STATUS)

    def queryDensity(self):
        msg = struct.pack('<B', 1)
        self.sendToBt(msg, BtCommandByte.PRT_GET_HEAT_DENSITY)

    def sendFeedToHeadLineToBt(self, length):
        msg = struct.pack('<H', length)
        self.sendToBt(msg, BtCommandByte.PRT_FEED_TO_HEAD_LINE)

    def queryPowerOffTime(self):
        msg = struct.pack('<B', 1)
        self.sendToBt(msg, BtCommandByte.PRT_GET_POWER_DOWN_TIME)

    def querySNFromBt(self):
        msg = struct.pack('<B', 1)
        self.sendToBt(msg, BtCommandByte.PRT_GET_SN)

    def queryHardwareInfo(self):
        msg = struct.pack('<B', 1)
        self.sendToBt(msg, BtCommandByte.PRT_GET_HW_INFO)