mirror of
https://github.com/NaitLee/Cat-Printer.git
synced 2025-05-16 07:10:30 -07:00
Possible GB03 support, i18n for CLI (optional)
This commit is contained in:
parent
6d86abbae6
commit
592ee8b9f7
2
.gitignore
vendored
2
.gitignore
vendored
@ -22,6 +22,8 @@ build-common/bleak
|
||||
build-common/python-win32*
|
||||
# dev config
|
||||
config.json
|
||||
# traffic dump
|
||||
*.dump
|
||||
# some other junk
|
||||
.directory
|
||||
thumbs.db
|
||||
|
10
.vscode/launch.json
vendored
10
.vscode/launch.json
vendored
@ -4,6 +4,14 @@
|
||||
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
|
||||
"version": "0.2.0",
|
||||
"configurations": [
|
||||
{
|
||||
"name": "Python: Current File",
|
||||
"type": "python",
|
||||
"request": "launch",
|
||||
"program": "${file}",
|
||||
"console": "integratedTerminal",
|
||||
"justMyCode": true
|
||||
},
|
||||
{
|
||||
"name": "Debug Web Interface",
|
||||
"type": "python",
|
||||
@ -14,4 +22,4 @@
|
||||
"justMyCode": true
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
@ -74,7 +74,7 @@ For all supported platforms,
|
||||
You can also use "pure" edition once you have [Python 3](https://www.python.org/) installed,
|
||||
or "bare" edition if you also managed to install `bleak` via `pip`.
|
||||
|
||||
See the [releases](./releases) now! (`0.0.*` versions are legacy/deprecated)
|
||||
See the [releases](https://github.com/NaitLee/Cat-Printer/releases) now!
|
||||
|
||||
## Problems?
|
||||
|
||||
|
64
i18n.py
Normal file
64
i18n.py
Normal file
@ -0,0 +1,64 @@
|
||||
'Minimal internationalization lib'
|
||||
|
||||
import os
|
||||
import math
|
||||
import json
|
||||
import locale
|
||||
|
||||
class I18n():
|
||||
''' Minimal implementation of current frontend i18n in Python
|
||||
Not Complete (yet)!
|
||||
'''
|
||||
|
||||
lang: str
|
||||
fallback: str
|
||||
data: dict = {
|
||||
'values': {},
|
||||
'contexts': []
|
||||
}
|
||||
|
||||
def __init__(self, search_path='lang', lang=None, fallback=None):
|
||||
self.lang = lang or locale.getdefaultlocale()[0]
|
||||
self.fallback = fallback or 'en_US'
|
||||
self.load_file(os.path.join(search_path, self.fallback.replace('_', '-') + '.json'))
|
||||
for name in os.listdir(search_path):
|
||||
if name == self.lang.replace('_', '-') + '.json':
|
||||
self.load_file(os.path.join(search_path, name))
|
||||
|
||||
def load_file(self, name):
|
||||
with open(name, 'r', encoding='utf-8') as file:
|
||||
self.load_data(file.read())
|
||||
|
||||
def load_data(self, raw_json):
|
||||
data = json.loads(raw_json)
|
||||
for key in data['values']:
|
||||
self.data['values'][key] = data['values'][key]
|
||||
if data.get('contexts') is not None:
|
||||
self.data['contexts'] = data['contexts']
|
||||
|
||||
def __getitem__(self, keys):
|
||||
if not isinstance(keys, tuple):
|
||||
keys = (keys, )
|
||||
data = self.data['values'].get(keys[0], keys[0])
|
||||
string = data[0][2] if isinstance(data, list) else data
|
||||
for i in keys:
|
||||
if isinstance(i, (int, float)):
|
||||
if string is None:
|
||||
string = data
|
||||
if isinstance(data, list):
|
||||
for j in data:
|
||||
if j[0] is None:
|
||||
j[0] = -math.inf
|
||||
if j[1] is None:
|
||||
j[1] = math.inf
|
||||
if j[0] < i < j[1]:
|
||||
template = j[2]
|
||||
break
|
||||
string = template.replace('%%n', i).replace('-%%n', -i)
|
||||
elif isinstance(i, dict):
|
||||
# not verified if would work
|
||||
if string is None:
|
||||
string = data
|
||||
for j in i:
|
||||
string = string.replace('%%{%s}' % j, i[j])
|
||||
return string
|
121
printer.py
121
printer.py
@ -4,13 +4,27 @@ import io
|
||||
import sys
|
||||
import argparse
|
||||
import asyncio
|
||||
from typing import List, Union, Any, Mapping
|
||||
from bleak import BleakClient, BleakScanner
|
||||
from bleak.exc import BleakError, BleakDBusError
|
||||
|
||||
try:
|
||||
from i18n import I18n
|
||||
except ImportError:
|
||||
class I18n():
|
||||
def __init__(self, _search_path=None, _lang=None, _fallback=None):
|
||||
pass
|
||||
def __getitem__(self, keys):
|
||||
if not isinstance(keys, tuple):
|
||||
keys = (keys, )
|
||||
return ' '.join([str(x) for x in keys])
|
||||
|
||||
i18n = I18n('www/lang')
|
||||
|
||||
class PrinterError(Exception):
|
||||
'Error of Printer driver'
|
||||
|
||||
models = ('GB01', 'GB02', 'GT01')
|
||||
models = ('GT01', 'GB01', 'GB02', 'GB03')
|
||||
|
||||
crc8_table = [
|
||||
0x00, 0x07, 0x0e, 0x09, 0x1c, 0x1b, 0x12, 0x15, 0x38, 0x3f, 0x36, 0x31,
|
||||
@ -38,7 +52,7 @@ crc8_table = [
|
||||
]
|
||||
|
||||
|
||||
def crc8(data):
|
||||
def crc8(data: Union[bytes, bytearray]):
|
||||
'crc8 hash'
|
||||
crc = 0
|
||||
for byte in data:
|
||||
@ -46,7 +60,7 @@ def crc8(data):
|
||||
return crc & 0xFF
|
||||
|
||||
|
||||
def set_attr_if_not_none(obj, attrs):
|
||||
def set_attr_if_not_none(obj: Any, attrs: Mapping[str, str]):
|
||||
''' set the attribute of `obj` if the value is not `None`
|
||||
`attrs` is `dict` of attr-value pair
|
||||
'''
|
||||
@ -56,20 +70,26 @@ def set_attr_if_not_none(obj, attrs):
|
||||
setattr(obj, name, value)
|
||||
|
||||
|
||||
def reverse_binary(value):
|
||||
def reverse_binary(value: int):
|
||||
'Get the binary value of `value` and return the binary-reversed form of it as an `int`'
|
||||
return int(f"{bin(value)[2:]:0>8}"[::-1], 2)
|
||||
|
||||
|
||||
def make_command(command, payload):
|
||||
'Make a `bytes` with command data, which can be sent to printer directly to operate'
|
||||
def make_command(command: int, payload: Union[bytes, bytearray], *, prefix: List[int]=None) -> bytearray:
|
||||
'Make a `bytearray` with command data, which can be sent to printer directly to operate'
|
||||
if len(payload) > 0x100:
|
||||
raise Exception('Too large payload')
|
||||
message = bytearray([0x51, 0x78, command, 0x00, len(payload), 0x00])
|
||||
message += payload
|
||||
message.append(crc8(payload))
|
||||
message.append(0xFF)
|
||||
return bytes(message)
|
||||
message = bytearray()
|
||||
if prefix is not None:
|
||||
message += prefix
|
||||
message += bytearray([
|
||||
0x51, 0x78,
|
||||
command, 0x00,
|
||||
len(payload), 0x00,
|
||||
*payload, crc8(payload),
|
||||
0xFF
|
||||
])
|
||||
return message
|
||||
|
||||
|
||||
class PrinterCommands():
|
||||
@ -97,6 +117,7 @@ class PBMData():
|
||||
self.height = height
|
||||
self.data = data
|
||||
self.args = {
|
||||
# setting to \x01 may make it faster. but don't know if there are drawbacks
|
||||
PrinterCommands.DrawingMode: b'\x00',
|
||||
PrinterCommands.SetEnergy: b'\xE0\x2E',
|
||||
PrinterCommands.SetQuality: b'\x05'
|
||||
@ -109,6 +130,12 @@ class PBMData():
|
||||
class PrinterDriver():
|
||||
'Manipulator of the printer'
|
||||
|
||||
name: str = None
|
||||
'The Bluetooth name of the printer'
|
||||
|
||||
address: str = None
|
||||
'The Bluetooth MAC address of the printer'
|
||||
|
||||
frequency = 0.8
|
||||
''' Time to wait between communication to printer, in seconds,
|
||||
too low value will cause gaps/tearing of printed content,
|
||||
@ -167,6 +194,10 @@ class PrinterDriver():
|
||||
|
||||
def _pbm_data_to_raw(self, data: PBMData):
|
||||
buffer = bytearray()
|
||||
# new/old print command
|
||||
if self.name == 'GB03':
|
||||
buffer.append(0x12)
|
||||
buffer += bytearray([0x51, 0x78, 0xa3, 0x00, 0x01, 0x00, 0x00, 0x00, 0xff])
|
||||
for key in data.args:
|
||||
buffer += make_command(key, data.args[key])
|
||||
buffer += make_command(
|
||||
@ -187,11 +218,11 @@ class PrinterDriver():
|
||||
)
|
||||
# buffer += make_command(
|
||||
# PrinterCommands.UpdateDevice,
|
||||
# bytes([0x00])
|
||||
# bytearray([0x00])
|
||||
# )
|
||||
buffer += make_command(
|
||||
PrinterCommands.DrawBitmap,
|
||||
bytes([reverse_binary(x) for x in data_for_a_line])
|
||||
bytearray([reverse_binary(x) for x in data_for_a_line])
|
||||
)
|
||||
buffer += make_command(
|
||||
PrinterCommands.LatticeControl,
|
||||
@ -201,12 +232,13 @@ class PrinterDriver():
|
||||
if self.feed_after > 0:
|
||||
buffer += make_command(
|
||||
PrinterCommands.FeedPaper,
|
||||
bytes([self.feed_after % 256, self.feed_after // 256])
|
||||
bytearray([self.feed_after % 256, self.feed_after // 256])
|
||||
)
|
||||
return buffer
|
||||
|
||||
async def send_buffer(self, buffer: bytearray, address: str):
|
||||
async def send_buffer(self, buffer: bytearray, address: str=None):
|
||||
'Send manipulation data (buffer) to the printer via bluetooth'
|
||||
address = address or self.address
|
||||
client = BleakClient(address, timeout=5.0)
|
||||
await client.connect()
|
||||
count = 0
|
||||
@ -224,7 +256,7 @@ class PrinterDriver():
|
||||
break
|
||||
await client.disconnect()
|
||||
|
||||
async def search_all_printers(self, timeout):
|
||||
async def search_all_printers(self, timeout: int):
|
||||
''' Search for all printers around with bluetooth.
|
||||
Only known-working models will show up.
|
||||
'''
|
||||
@ -235,7 +267,7 @@ class PrinterDriver():
|
||||
if device.name in models:
|
||||
result.append(device)
|
||||
return result
|
||||
async def search_printer(self, timeout):
|
||||
async def search_printer(self, timeout: int):
|
||||
'Search for a printer, returns `None` if not found'
|
||||
timeout = timeout or 3
|
||||
devices = await self.search_all_printers(timeout)
|
||||
@ -243,14 +275,16 @@ class PrinterDriver():
|
||||
return devices[0]
|
||||
return None
|
||||
|
||||
async def print_file(self, path: str, address: str):
|
||||
async def print_file(self, path: str, address: str=None):
|
||||
'Method to print the specified PBM image at `path` with printer at specified MAC `address`'
|
||||
address = address or self.address
|
||||
pbm_data = self._read_pbm(path)
|
||||
buffer = self._pbm_data_to_raw(pbm_data)
|
||||
await self.send_buffer(buffer, address)
|
||||
|
||||
async def print_data(self, data: bytes, address: str):
|
||||
async def print_data(self, data: bytes, address: str=None):
|
||||
'Method to print the specified PBM image `data` with printer at specified MAC `address`'
|
||||
address = address or self.address
|
||||
pbm_data = self._read_pbm(None, data)
|
||||
buffer = self._pbm_data_to_raw(pbm_data)
|
||||
await self.send_buffer(buffer, address)
|
||||
@ -259,46 +293,47 @@ class PrinterDriver():
|
||||
async def _main():
|
||||
'Main routine for direct command line execution'
|
||||
parser = argparse.ArgumentParser(
|
||||
description='''Print an PBM image to a Cat/Kitty Printer, of model GB01, GB02 or GT01.'''
|
||||
description=' '.join([
|
||||
i18n['print-pbm-image-to-cat-printer'],
|
||||
i18n['supported-models-'],
|
||||
str(models)
|
||||
])
|
||||
)
|
||||
parser.add_argument('file', default='-', metavar='FILE', type=str,
|
||||
help='PBM image file to print, use \'-\' to read from stdin')
|
||||
help=i18n['path-to-pbm-file-dash-for-stdin'])
|
||||
exgr = parser.add_mutually_exclusive_group()
|
||||
exgr.add_argument('-s', '--scan', metavar='DELAY', default=3.0, required=False, type=float,
|
||||
help='Scan for a printer for specified seconds')
|
||||
exgr.add_argument('-s', '--scan', metavar='TIME', default=3.0, required=False, type=float,
|
||||
help=i18n['scan-for-specified-seconds'])
|
||||
exgr.add_argument('-a', '--address', metavar='xx:xx:xx:xx:xx:xx', required=False, type=str,
|
||||
help='The printer\'s bluetooth MAC address')
|
||||
parser.add_argument('-p', '--feed', required=False, type=int,
|
||||
help='Extra paper to feed after printing')
|
||||
parser.add_argument('-f', '--freq', required=False, type=float,
|
||||
help='Communication frequency, in seconds. ' +
|
||||
'set a bit higher (eg. 1 or 1.2) if printed content is teared/have gaps')
|
||||
help=i18n['specify-printer-mac-address'])
|
||||
parser.add_argument('-f', '--freq', metavar='FREQ', required=False, type=float,
|
||||
help=i18n['communication-frequency-0.8-or-1-recommended'])
|
||||
parser.add_argument('-d', '--dry', required=False, action='store_true',
|
||||
help='Emulate the printing process, but actually print nothing ("dry run")')
|
||||
parser.add_argument('-m', '--mtu', required=False, type=int,
|
||||
help='MTU of bluetooth packet (Advanced)')
|
||||
help=i18n['dry-run-test-print-process-only'])
|
||||
cmdargs = parser.parse_args()
|
||||
addr = cmdargs.address
|
||||
printer = PrinterDriver()
|
||||
if not addr:
|
||||
print('Cat Printer :3')
|
||||
print(f' * Finding printer devices via bluetooth in {cmdargs.scan} seconds')
|
||||
print(i18n['cat-printer'])
|
||||
print(i18n['scanning-for-devices'])
|
||||
device = await printer.search_printer(cmdargs.scan)
|
||||
if device is not None:
|
||||
print(f' * Will print through {device.name} {device.address}')
|
||||
print(i18n['printing'])
|
||||
else:
|
||||
print(' ! No device found. Please check if the printer is powered on.')
|
||||
print(' ! Or try to scan longer with \'-s 6.0\'')
|
||||
print(i18n['no-available-devices-found'])
|
||||
print(i18n['please-check-if-the-printer-is-down'])
|
||||
print(i18n['or-try-to-scan-longer'])
|
||||
sys.exit(1)
|
||||
if cmdargs.dry:
|
||||
print(' * DRY RUN')
|
||||
print(i18n['dry-run'])
|
||||
set_attr_if_not_none(printer, {
|
||||
'feed_after': cmdargs.feed,
|
||||
'name': device.name,
|
||||
'address': device.address,
|
||||
'frequency': cmdargs.freq,
|
||||
'mtu': cmdargs.mtu,
|
||||
'dry': cmdargs.dry
|
||||
'dry_run': cmdargs.dry
|
||||
})
|
||||
await printer.print_file(cmdargs.file, addr)
|
||||
await printer.print_file(cmdargs.file)
|
||||
print(i18n['finished'])
|
||||
|
||||
async def main():
|
||||
'Run the `_main` routine while catching exceptions'
|
||||
@ -311,7 +346,7 @@ async def main():
|
||||
(isinstance(e, BleakDBusError) and
|
||||
getattr(e, 'dbus_error') == 'org.bluez.Error.NotReady')
|
||||
):
|
||||
print(' ! Please enable bluetooth on this machine :3')
|
||||
print(i18n['please-enable-bluetooth'])
|
||||
sys.exit(1)
|
||||
else:
|
||||
raise
|
||||
|
@ -58,11 +58,10 @@ class PrinterServer(BaseHTTPRequestHandler):
|
||||
'(Local) server for Cat Printer Web interface'
|
||||
buffer = 4 * 1024 * 1024
|
||||
max_payload = buffer * 16
|
||||
printer_address: str = None
|
||||
settings = DictAsObject({
|
||||
'config_path': 'config.json',
|
||||
'is_android': False,
|
||||
'printer_address': None,
|
||||
'printer': None,
|
||||
'scan_time': 3,
|
||||
'frequency': 0.8,
|
||||
'dry_run': False
|
||||
@ -133,15 +132,16 @@ class PrinterServer(BaseHTTPRequestHandler):
|
||||
body = self.rfile.read(content_length)
|
||||
api = self.path[1:]
|
||||
if api == 'print':
|
||||
if self.settings.printer_address is None:
|
||||
if self.settings.printer is None:
|
||||
# usually can't encounter, though
|
||||
raise PrinterServerError('No printer address specified')
|
||||
Printer.dry_run = self.settings.dry_run
|
||||
Printer.frequency = float(self.settings.frequency)
|
||||
Printer.name, Printer.address = self.settings.printer.split(',')
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
devices = loop.run_until_complete(
|
||||
Printer.print_data(body, self.settings.printer_address)
|
||||
Printer.print_data(body)
|
||||
)
|
||||
self.api_success()
|
||||
finally:
|
||||
|
@ -51,6 +51,14 @@
|
||||
"you-can-close-this-page-manually": "You can close this page manually",
|
||||
"please-enable-bluetooth": "Please enable Bluetooth",
|
||||
"error-happened-please-check-error-message": "Error happened, please check error message",
|
||||
"you-can-seek-for-help-with-detailed-info-below": "You can seek for help with detailed info below."
|
||||
"you-can-seek-for-help-with-detailed-info-below": "You can seek for help with detailed info below.",
|
||||
|
||||
"or-try-to-scan-longer": "Or try to scan longer",
|
||||
"print-pbm-image-to-cat-printer": "Print PBM image to Cat Printer.",
|
||||
"supported-models-": "Supported models:",
|
||||
"path-to-pbm-file-dash-for-stdin": "Path to PBM file. '-' for stdin.",
|
||||
"scan-for-specified-seconds": "Scan for specified seconds",
|
||||
"specify-printer-mac-address": "Specify printer MAC address",
|
||||
"communication-frequency-0.8-or-1-recommended": "Communication frequency. 0.8 or 1 recommended."
|
||||
}
|
||||
}
|
@ -50,6 +50,14 @@
|
||||
"you-can-close-this-page-manually": "您可手动关闭此页面",
|
||||
"please-enable-bluetooth": "请启用蓝牙",
|
||||
"error-happened-please-check-error-message": "发生错误,请检查错误消息",
|
||||
"you-can-seek-for-help-with-detailed-info-below": "您可以使用以下详细信息寻求帮助。"
|
||||
"you-can-seek-for-help-with-detailed-info-below": "您可以使用以下详细信息寻求帮助。",
|
||||
|
||||
"or-try-to-scan-longer": "或者尝试延长扫描时间",
|
||||
"print-pbm-image-to-cat-printer": "打印 PBM 图片到猫咪打印机。",
|
||||
"supported-models-": "支持的型号:",
|
||||
"path-to-pbm-file-dash-for-stdin": "PBM 文件的位置。“-” 作为标准输入。",
|
||||
"scan-for-specified-seconds": "扫描指定的时长。",
|
||||
"specify-printer-mac-address": "指定打印机的 MAC 地址",
|
||||
"communication-frequency-0.8-or-1-recommended": "通讯频率。推荐 0.8 或 1。"
|
||||
}
|
||||
}
|
@ -374,7 +374,7 @@ class Main {
|
||||
putEvent('#button-print', 'click', this.print, this);
|
||||
putEvent('#device-refresh', 'click', this.searchDevices, this);
|
||||
this.attachSetter('#scan-time', 'change', 'scan_time');
|
||||
this.attachSetter('#device-options', 'input', 'printer_address');
|
||||
this.attachSetter('#device-options', 'input', 'printer');
|
||||
this.attachSetter('input[name="algo"]', 'change', 'mono_algorithm');
|
||||
this.attachSetter('#transparent-as-white', 'change', 'transparent_as_white');
|
||||
this.attachSetter('#dry-run', 'change', 'dry_run',
|
||||
@ -482,7 +482,7 @@ class Main {
|
||||
let search_result = await callApi('/devices', null, this.bluetoothProblemHandler);
|
||||
if (search_result === null) return;
|
||||
let devices = search_result.devices;
|
||||
this.deviceOptions.childNodes.forEach(e => e.remove());
|
||||
[... this.deviceOptions.children].forEach(e => e.remove());
|
||||
if (devices.length === 0) {
|
||||
Notice.notice('no-available-devices-found');
|
||||
hint('#device-refresh');
|
||||
@ -492,8 +492,8 @@ class Main {
|
||||
hint('#insert-picture');
|
||||
devices.forEach(device => {
|
||||
let option = document.createElement('option');
|
||||
option.value = device.address;
|
||||
option.innerText = `${device.name}-${device.address.slice(12, 14)}${device.address.slice(15)}`;
|
||||
option.value = `${device.name},${device.address}`;
|
||||
option.innerText = `${device.name}-${device.address.slice(3, 5)}${device.address.slice(0, 2)}`;
|
||||
this.deviceOptions.appendChild(option);
|
||||
});
|
||||
this.deviceOptions.dispatchEvent(new Event('input'));
|
||||
|
Loading…
x
Reference in New Issue
Block a user