逆向某错题打印机，实现自定义内容流式打印

啵哩口袋打印机（Poooli）是那种巴掌大、用热敏纸、靠蓝牙连接的口袋打印机，它官方只有手机 App，而我们都知道它的公司已经挂注销状态了，很明显印不了我自己的东西。所以我想把它接管过来：让电脑里的图片、文字，甚至一个 SSH 会话，都直接变成纸上的墨点

不过，这套协议没有公开文档，格式是一点点抓包猜出来的；下面先讲那段经历——怎么抓包、怎么撞协议、怎么验证猜想；挖出来的格式再单独整理一遍；最后看它在代码里长成什么样

0x00 逆向：协议是怎么抓出来的

遇事不决先逆向，看看这小东西是怎么实现的。

它到底是什么设备

协议长什么样，其实是硬件定死的：我的啵哩 L2 是一台 203dpi 的热敏打印机，不用墨水，靠加热头把纸烧黑，打印区固定 640 像素宽；这两个数字很关键——热敏意味着只有黑白两态，640 意味着每行的字节数是写死的

蓝牙那边走的是经典蓝牙，不是 BLE，上面跑 RFCOMM —— 这是一种串口仿真协议，默认 channel 1，这一点决定了协议的整体形态：没有 GATT 特征值可以寻址，「发一条指令」在物理层就是「往串口里写一串字节」，成帧的活儿得协议自己干

合起来看，它就是一台挂在蓝牙串口上、只会按行吃点阵的黑盒

从抓包开始

啵哩没有公开协议，官方 App 是唯一「会说话」的客户端；想让它说给我听，就得在它和打印机之间架一个中转站，整个过程分五步：录下来 → 剥到裸字节 → 认出握手 → 解开长度字段 → 认出压缩算法，下面按这个顺序来

第一步：想办法录下来

蓝牙 RFCOMM 是串口风格的字节流，Linux 上让 BlueZ 把 HCI 流量吐出来：

# 方式一：btmon 直接写 btsnoop，Wireshark 原生支持
sudo btmon --write ~/poooli.btsnoop

# 方式二：hcidump 打原始 hex，适合不喜欢 GUI 的 Geek
sudo hcidump --raw -i hci0 > poooli.raw

让蓝牙适配器进入监听模式的前提是它本身在工作（hciconfig 里能看到 UP RUNNING）；如果手上只有 USB 蓝牙棒，也可以直接插一台能跑 BlueZ 的机器当中间人，让手机连它

录的时候要做的动作很固定，一次录一件事，方便后面按动作对齐：

1. 打开官方 App，点「连接设备」
2. 新建一张纯白图片，打印
3. 新建一张左半黑、右半白的图片，打印
4. 输入一行短文字，打印

四次动作录进四个文件，为什么要四份：纯白的用来找「压缩后极短」的行，左半黑的用来确定位序，文字的用来确认 ASCII 指令——对照组是逆向里最省时间的东西

第二步：剥到裸字节

.btsnoop 用 Wireshark 打开，从 HCI ACL → L2CAP → RFCOMM 一层层点进去；手动看太累，用 tshark 把 RFCOMM 的载荷直接导出来：

# 把 RFCOMM payload 导出成一行一串 hex
tshark -r ~/poooli.btsnoop \
       -Y 'btbrfcomm' \
       -T fields -e btbrfcomm.data 2>/dev/null \
  | tr -d ':' > frames.hex

（不同版本的 Wireshark 字段名可能略有差别，-G fields | grep rfcomm 查一下就有了）

导出来是一坨连续的 hex，写个小脚本把它还原成分帧的样子：

"""把 tshark 导出的 RFCOMM 载荷画成一棵「长度树」，先找固定前缀。"""

from collections import Counter
from pathlib import Path

raw = bytes.fromhex(Path("frames.hex").read_text().strip())

# 统计所有 4 字节窗口，出现次数远高于平均的那些就是候选帧头。
counts = Counter(raw[i:i + 4] for i in range(len(raw) - 4))
for window, count in counts.most_common(12):
    print(count, window.hex(" "))

第一次跑的输出（节选）大概是这样的：

41 10 7b 3d 3d
2  1b 1c 73 65
2  10 7e 68 79
...

10 7b 3d 3d 出现了 41 次，远超其它窗口——而纯白图正好 41 行，第一个结论立刻成立：10 7b 3d 3d 是每一行图像数据的开头

第三步：认出握手

另外两个高频窗口 1b 1c 73 65 和 10 7e 68 79 只出现两次（四次动作各一次连接），说明它们是每次连接发一次的握手帧，把完整的几组抓出来对齐：

1b 1c 73 65 74 20 6d 6d 05 08      10 7e 68 79 7a ed 09
└────────────── 10 字节 ───────┘    └───── 7 字节 ─────┘

1b 1c 73 65 74 20 6d 6d 一眼可读：ESC FS + ASCII "set mm"；PAPER_WIDTH 那条则完全是魔数，但它是定长、每次都在第二位、和「纸」有关——按位置和出现规律命名就够了

逆向时不用急着弄懂每条指令的含义，按「出现次数 + 位置」先起个工作名就行，名字写在注释里，等后面用代码验证过再定稿

第四步：解开长度字段

接下来是硬骨头，把每一行的信封切出来：

10 7b 3d 3d   a1 a2   b1 b2   c1 c2 c3 c4   <数据...>

一行数据的边界靠 10 7b 3d 3d 就能切，切完之后，观察后面那三个「长度」：

• a 在纯白图上几乎不变；
• b 恒等于某个值；
• c 每行都不同，而且和它后面那段数据的长度对不上——差一个固定关系

先量一下 c 后面实际的字节数，再和 c 的值放在一起看：

"""把每行的三个长度字段和真实数据长度列出来，找异或关系。"""

import struct
from pathlib import Path

raw = bytes.fromhex(Path("frames.hex").read_text().strip())
HEAD = bytes.fromhex("107b3d3d")

offset = 0
rows = []
while True:
    start = raw.find(HEAD, offset)
    if start < 0:
        break
    a, b, c = struct.unpack_from("<HHI", raw, start + 4)
    rows.append((a, b, c))
    offset = start + 12          # 先按 12 字节头跳，数据部分靠人工对齐

for a, b, c in rows[:8]:
    print("a=%#06x b=%#06x c=%#010x" % (a, b, c))

输出（节选，纯白图的几行）：

a=0x0d5d b=0x0d0c c=0x0d0d0d1f
a=0x0d5d b=0x0d0c c=0x0d0d0d14
a=0x0d5d b=0x0d0c c=0x0d0d0d1a

a 和 b 是定值，c 的高两位 0d 0d 也几乎不动——这就是掩码的味道，拿几个已知值试异或：

for mask in (0x0d, 0x0d0d, 0x0d0d0d0d):
    print(hex(mask), [hex(v ^ mask) for v in (0x0d5d, 0x0d0c, 0x0d0d0d1f)])

0xd      [0xd50, 0xd01, 0xd0d0d12]
0xd0d    [0x50, 0x1, 0x0d0d0012]
0xd0d0d0 [0xd50, 0xd01, 0x12]

第二列是最漂亮的：0x0d5d ^ 0x0d0d = 0x50 = 80，0x0d0c ^ 0x0d0d = 1，一个正好是 640 / 8，一个正好是「一行」——掩码就是 0x0d0d，长度字段用它，四字节长度字段用 0x0d0d0d0d

结论立刻闭环：三个字段 = 宽度 ^ 0x0d0d、行数 ^ 0x0d0d、压缩长度 ^ 0x0d0d0d0d

顺便，c 解开之后就能精确切出每帧的数据段了——把循环里的 offset 从「+12 硬跳」改成「按长度跳」，整段流可以完整解完：

offset = start + 12
length = c ^ 0x0D0D0D0D
payload = raw[offset:offset + length]
offset += length

第五步：认出压缩算法

payload 现在是一段二进制，它有几个很明显的特征：

• 纯白行的 payload 只有 2~6 个字节（因为 80 个 0x00 被压没了）；
• 有内容的行明显变长，但和「原始 80 字节」没有线性关系；
• 开头几个字节反复出现 f0、13、40 这类值

这是压缩的典型特征：冗余越多，输出越短——加密不会这样

拿压缩库一个个试，第一个想到的是 zlib，因为 Python 自带：

import zlib
for data in payloads:
    try:
        print(zlib.decompress(data ^ 0x0d))
    except zlib.error as error:
        print("不是 zlib:", error)

全军覆没，第二个试 LZO——小票机、打印机、嵌入式设备上用 LZO 的非常多：

import lzo

row = bytes(b ^ 0x0D for b in payload)     # 先把那层异或剥掉
print(lzo.decompress(row, False))          # False = 不带头部

一行 80 字节的原始数据原样还原，和「左半黑右半白」的测试图对得上：

b'\xff' * 40 + b'\x00' * 40

（0xff = 8 个黑点，40 字节 = 左半 320 像素黑；右半全白 0x00）；位序、黑点、压缩算法、异或掩码，一次全部验证通过

把结论固化成一张对照表

逆向做完，把每一步的猜测和验证结果整理成表，这张表后来直接变成了代码里的常量：

这张表就是协议文档的初稿——代码里的四行常量、一个信封格式，全部来自它

抓包阶段的三个坑

当时踩过的坑比结论更值得记一笔：

1. 不要相信「长度字段看起来像长度」：第一版我直接按 c 的值去切数据，结果每切一行错位一格，整段流越解越乱；后来意识到 c 被异或过，用掩码解出来才对齐
2. 不要跳过握手：握手两条指令和图像帧的编码风格不一样（握手是字面量，图像帧是长度 + 压缩），如果只盯着图像帧，会以为协议全是二进制的
3. 对照组比猜测快十倍：纯白图、半黑图、文字，三张图一对比，压缩率、位序、行数三个问题一次性排掉；这比反复对着 App 点「打印」猜要高效得多

到这里，格式已经齐了，下面把它整理成一份干净的说明

0x01 协议

抓包的结论整理出来，就是下面这套格式；它算不上复杂：四种帧、一个信封格式、一层异或，再加一次 LZO 压缩

四种帧

把整段流量按功能拆开，一共就四种帧：

INITIALIZE 一眼能认出一半：1b 1c 是 ESC FS（小票外设常用的转义前缀），后面干脆是字面量 set mm——73 65 74 20 6d 6d 就是 ASCII 的 "set mm"，这类「半明文 + 半魔数」的指令在这个协议里到处都是

在代码里它们就是四个常量，没有任何解释层：

class Poooli:
    INITIALIZE = b"\x1b\x1cset mm\x05\x08"
    PAPER_WIDTH = b"\x10\x7e\x68\x79\x7a" + b"\xed\x09"
    BMP = b"\x10\x7b\x3d\x3d"
    LINE_FEED = b"\x16\x16\x0c\x57\x0d"
    WIDTH = 640

连接建立后的握手就是「把前两条按顺序发出去」，仅此而已：

def connect(self, mac, channel=1):
    LOG.debug("socket.connect(%s, %d)", mac, channel)
    started = time.monotonic()
    self.socket.connect((mac, channel))
    LOG.info("已连接，用时 %.2f 秒", time.monotonic() - started)
    self.socket.send(Poooli.INITIALIZE)
    self.socket.send(Poooli.PAPER_WIDTH)
    LOG.debug("握手已发送：%s", (Poooli.INITIALIZE + Poooli.PAPER_WIDTH).hex(" "))

注意 connect 里没有任何「等应答」的逻辑，这也是抓包看出来的：打印机对握手不回复，它是单向的，发完就认

每一行数据的信封

图像是按行发的，一行一个信封，格式固定：

10 7b 3d 3d   5d 0d   0c 0d   ZZ ZZ ZZ ZZ   <压缩数据>
└─ BMP 头 ──┘  └宽度┘  └行数┘  └─ 压缩长度 ─┘

四段的含义：

• 头：10 7b 3d 3d，就是 BMP 常量
• 宽度：本行有多少字节（= 纸宽 / 8 = 80），异或 0x0d0d 之后变成 5d 0d
• 行数：固定 1，异或 0x0d0d 之后是 0c 0d，必须逐行发
• 压缩长度：这一行压缩后的字节数，异或 0x0d0d0d0d
• 数据：LZO1X-1 压缩后的裸块（不带头部），每个字节再异或 0x0d

三个长度字段全部用 0x0d 重复填满做异或，这是协议里唯一的「混淆层」；它不是加密，更像随手加的偏置——但没猜到它，抓包看到的长度字段永远是乱的

一行到底多少字节，这个数必须算明白

640 像素、每字节 8 个像素，所以 640 / 8 = 80，但协议文档（包括这个仓库早期的 docs/poooli.md）里写着「每行 48 字节」——这是错的；48 是抓包时常见的压缩后长度，被误当成了宽度字段；真正的宽度字段是 80，异或 0x0d0d 之后是 5d 0d：

80        = 0x0050
0x0050 ^ 0x0d0d = 0x0d5d   → 小端两根字节：5d 0d
1         = 0x0001
0x0001 ^ 0x0d0d = 0x0d0c   → 0c 0d

所以每一行的信封实际是：

10 7b 3d 3d   5d 0d   0c 0d   ZZ ZZ ZZ ZZ   <压缩数据>
└─ BMP 头 ──┘  └宽度┘  └行数┘  └─ 压缩长度 ─┘
宽度 = 80 ^ 0x0d0d = 0x0d5d
行数 = 1  ^ 0x0d0d = 0x0d0c

有个地方容易看走眼：协议里的「宽度」说的是这一行的字节数，它恰好等于 纸宽 / 8；两者混起来看很容易把常量写错，我在最初那版里就照着那段文档把 48 写死了，结果打印机接收帧后错位、整张图扭曲；后来改成从 Poooli.WIDTH // 8 算，这个 bug 才彻底消失

位序与黑点

位图打包的规则有两条，都必须和打印机一致：

• MSB first：一个字节里的 8 个像素，最左边的像素放在最高位
• 1 表示黑点：PIL 的二值图里黑色像素值是 0，所以这里要反过来判断

代码里就是这三行：

for bit in range(8):
    if image.getpixel((ix * 8 + bit, iy)) == 0:
        byte |= 1 << (7 - bit)

如果位序反了，印出来的就是一张镜像的乱码——能看出一点结构，但完全认不出内容；如果黑白反了，照片会变成底片；这两个错误在调试期各出现过一次，都是靠打印一张左边半黑、右边半白的测试图定位的

LZO：为什么是它，怎么用

点阵本身非常冗余：一整行 80 字节里，白纸部分全是 0x00；不做压缩的话，一张 640×800 的图就是 64KB 原始数据，走蓝牙要磨很久，所以协议用了 LZO1X-1

选择理由也很朴素：LZO 解压极快、内存占用小，非常适合单片机；打印机端解压，发送端压缩

关键是用哪种模式：LZO 有很多入口，协议要的是「裸块、无头部」的那一种：lzo1x_1_compress，python-lzo 里对应：

PYTHON_LZO.compress(image_bytes, 1, False)
#                                │  └─ False = 不带头部（裸块）
#                                └──── 1 = lzo1x_1

第三参数如果填 True，会在前面加一段 LZO 自己的头，打印机直接不认；这个参数也是抓包对比出来的：带头的输出比实际发出去的多几个字节，且首字节对不上

压缩完还要做两次异或：

lzno_xor = (len(lzno) ^ LENGTH_XOR).to_bytes(4, "little")   # 长度，小端 4 字节
image_lzo_xor = b"".join((b ^ BYTE_XOR).to_bytes(1, "little") for b in lzno)

LENGTH_XOR = 0x0D0D0D0D、BYTE_XOR = 13（也就是 0x0d），同一套掩码，发射顺序是「长度在前、数据在后」，长度本身也异或过

收尾

所有行发完后单独补一条 LINE_FEED（16 16 0c 57 0d）走纸：

self.socket.send(Poooli.LINE_FEED)
LOG.debug("送纸指令：%s", Poooli.LINE_FEED.hex(" "))

没有它，纸就停在打印头下面没切出来；多页打印时，每页末尾都会来一次，所以「页面数量」是可以直接数 LINE_FEED 的——测试就是这么数页数的

协议全景（一张图看完）

PC                                    啵哩打印机
│                                          │
├─ connect((mac, 1)) ──── RFCOMM ─────────►│
│                                          │
├─ INITIALIZE (1b 1c "set mm" 05 08) ─────►│   握手：不回复
├─ PAPER_WIDTH (10 7e 68 79 7a ed 09) ────►│
│                                          │
│   ┌─ 第 0 行 ────────────────────────┐   │
├───┤ 10 7b 3d 3d 5d 0d 0c 0d L L L L  ├──►│   80 字节点阵
│   │ <LZO(80B) ^ 0x0d，长度 ^ 0x0d0d0d0d>│   │
│   └──────────────────────────────────┘   │
│              ... 第 1..N-1 行 ...        │
│                                          │
├─ LINE_FEED (16 16 0c 57 0d) ────────────►│   走纸
│                                          │
├─ 关闭 socket ───────────────────────────►│

协议本身就这么点东西，四种帧、一个信封格式，抄在一张便签上都写得下；后面所有的复杂度，都在回答同一个问题：怎么把想印的内容变成这 80 字节一行

0x02 代码展示

下面把每一层真正落地的代码贴出来，顺序顺着数据流走：图片 → 位图 → LZO → 蓝牙帧 → 文本排版 → ssh 镜像 → CLI → 自检 → 打包 → CI

代码来自仓库，只做了必要的裁剪（省略与主题无关的分支）

0x03 协议层：组帧

poooli.py 是协议的唯一出口，核心是 _send_image：把一张二值图逐行打包、压缩、异或、发出去；这个函数是整个项目的「心脏」，也几乎是逆向结论的逐字翻译

def _send_image(self, image):
    width_bytes = Poooli.WIDTH // 8
    started = time.monotonic()
    raw_total = 0
    compressed_total = 0
    progress_step = max(1, image.height // 10)
    LOG.info("开始发送：%d 行 × %d 字节/行", image.height, width_bytes)
    for iy in range(image.height):
        image_bytes = b""
        for ix in range(width_bytes):
            byte = 0
            for bit in range(8):
                if image.getpixel((ix * 8 + bit, iy)) == 0:
                    byte |= 1 << (7 - bit)
            image_bytes += byte.to_bytes(1, "little")
        lzo_image, lzono = Poooli.compress_to_lzo(image_bytes)
        self.socket.send(Poooli.BMP)
        self.socket.send(
            (width_bytes ^ (LENGTH_XOR & 0xFFFF)).to_bytes(2, "little")
        )
        self.socket.send((1 ^ (LENGTH_XOR & 0xFFFF)).to_bytes(2, "little"))
        self.socket.send(lzono)
        self.socket.send(lzo_image)
        raw_total += len(image_bytes)
        compressed_total += len(lzo_image)
        if iy == 0:
            LOG.debug("第 1 行原始数据：%s", image_bytes.hex(" "))
        LOG.debug(
            "第 %d/%d 行：%d 字节 -> LZO %d 字节，发送 %d 段",
            iy + 1,
            image.height,
            len(image_bytes),
            len(lzo_image),
            5,
        )
        if (iy + 1) % progress_step == 0 and iy + 1 < image.height:
            LOG.info(
                "进度 %d%%（%d/%d 行，已发 %d 字节），用时 %.1f 秒",
                100 * (iy + 1) // image.height,
                iy + 1,
                image.height,
                compressed_total,
                time.monotonic() - started,
            )
    self.socket.send(Poooli.LINE_FEED)
    LOG.debug("送纸指令：%s", Poooli.LINE_FEED.hex(" "))
    LOG.info(
        "发送完成：%d 行，原始 %d 字节，LZO 后 %d 字节（%.0f%%），用时 %.1f 秒",
        image.height,
        raw_total,
        compressed_total,
        100.0 * compressed_total / raw_total if raw_total else 0.0,
        time.monotonic() - started,
    )

有三处值得琢磨：

• 每行单独 send 五段：RFCOMM 是字节流，多段 send 和拼成一段发完全等价，但分开写让日志和调试能对得上协议结构
• 进度日志按 10% 打：打印机很慢，没有进度会让人以为卡死了
• 统计压缩率：compressed_total / raw_total 是判断「是不是发了张全白的废纸」的第一手指标——纯白纸的压缩率会低到个位数百分比

压缩 + 异或是最小的一个函数：

@staticmethod
def compress_to_lzo(image_bytes):
    """LZO1X-1 compress one row and obfuscate it the way the printer wants.

    Returns the compressed bytes XORed with 0x0d and the compressed length
    XORed with 0x0d0d0d0d.
    """
    if PYTHON_LZO is not None:
        lzno = PYTHON_LZO.compress(image_bytes, 1, False)
    else:
        lzno = _compress_with_ctypes(image_bytes)
    lzno_xor = (len(lzno) ^ LENGTH_XOR).to_bytes(4, "little")
    image_lzo_xor = b""  # compressed data
    for ib in range(len(lzno)):
        image_lzo_xor += (lzno[ib] ^ BYTE_XOR).to_bytes(1, "little")
    return image_lzo_xor, lzno_xor

两个常量放在模块顶部，含义用注释锁死：

BYTE_XOR = 13                      # 0x0d，数据段逐字节异或
LENGTH_XOR = 0x0D0D0D0D            # 长度字段异或（低 2 字节给行宽/行数用）
WORK_MEMORY_SIZE = 16384 * 8       # ctypes 那条路要一块工作内存

0x04 图像层：把任何东西变成 640 宽二值图

协议只要一种输入：640 宽的单色位图，prepare_image 负责把任意 PIL 图片变成它

def prepare_image(
    self, image, mode="bnw_dither", contrast=1.0, brightness=1.0, rotate=True
):
    """把内存里的 PIL 图片处理成打印机要的 1 位位图。

    rotate=True 时横图会自动转 90°（照片的习惯）；文本排版渲染出来的位图
    传 rotate=False，否则整页文字会被竖过来。
    """
    LOG.debug("处理图片：%s %s", image.size, image.mode)
    image = image.convert("RGBA")
    if brightness != 1.0:
        image = ImageEnhance.Brightness(image).enhance(brightness)
        LOG.debug("亮度 ×%.2f", brightness)
    if contrast != 1.0:
        image = ImageEnhance.Contrast(image).enhance(contrast)
        LOG.debug("对比度 ×%.2f", contrast)
    if rotate and image.width > image.height:
        image = image.rotate(90, expand=True)
        LOG.debug("横图转 90°：%s", image.size)
    height = int(image.height * Poooli.WIDTH / image.width)
    if image.size != (Poooli.WIDTH, height):
        # 已经是对应宽度的位图（文本页就是）就直接用：Pillow 的 resize()
        # 即使尺寸不变也会重采样，会把已经二值化的点阵弄脏。
        image = image.resize(size=(Poooli.WIDTH, height))
        LOG.debug("等比缩放到 %s", image.size)
    else:
        LOG.debug("宽度已经是 %dpx，不缩放", Poooli.WIDTH)
    if mode == "bnw_dither":
        image = image.convert(mode="1")
    elif mode == "bnw":
        image = image.convert(mode="1", dither=Image.Dither.NONE)
    else:
        raise ValueError("Unknown mode: " + mode)
    LOG.info(
        "图像处理完成：%dx%d 单色位图（%s），黑色像素 %.1f%%",
        image.width,
        image.height,
        mode,
        Poooli.black_ratio(image),
    )
    return image

这里藏着四个「后来才补上」的细节：

1. 横图转 90°：照片是横的，纸是竖的；不转的话，一张 4:3 的照片会被压成一条细长的东西，人脸都认不出来；转过去之后再等比缩放到 640 宽，高度自动算
2. 已经是 640 宽就不 resize：Pillow 的 resize() 就算目标尺寸和当前一样，也会跑一遍重采样滤波器；对已经二值化的文本页来说，这一跑会把干净的边缘弄出灰边，再二值化就变成毛刺——所以加了个 image.size != (WIDTH, height) 的判断，文本页直接从这条路跳过
3. 两种二值化模式：bnw_dither 抖动，照片过渡自然；bnw 纯阈值，文字锐利
4. rotate 参数：文本页传 False，否则整页文字会被竖过来

黑点占比是个很实用的小工具，用来一眼判断「是不是整片糊黑 / 全白」：

@staticmethod
def black_ratio(image):
    """黑色像素占比，用来判断是不是整片糊黑 / 全白。"""
    total = image.width * image.height
    if not total:
        return 0.0
    histogram = image.histogram()
    return 100.0 * histogram[0] / total

对外暴露三个入口，覆盖「文件 / 内存图片 / 字节流」三种来源：

def send_image(self, file, mode="bnw_dither", contrast=1.0, brightness=1.0):
    self._send_image(self.process_image(file, mode, contrast, brightness))

def send_image_object(self, image, mode="bnw_dither", contrast=1.0, brightness=1.0):
    """发送内存里的 PIL 图片：文本排版渲染出来的位图走这条路。"""
    self._send_image(
        self.prepare_image(image, mode, contrast, brightness, rotate=False)
    )

def send_image_bytes(self, image_bytes, mode="bnw_dither", contrast=1.0, brightness=1.0):
    image = Image.open(io.BytesIO(image_bytes))
    self._send_image(self.prepare_image(image, mode, contrast, brightness))

0x05 LZO 双后端：让用户机器上「什么都没装」也能跑

这是我在打包阶段花时间最多的一块，目标很明确：打出来的程序，用户双击就能用，不需要装 python-lzo，也不需要装 liblzo2

策略是两条路：

1. python-lzo 存在就用它（Windows 轮子自带静态 LZO，Linux 上编译时链系统 liblzo2）；
2. 没有就退回 ctypes 动态加载系统库

try:
    # python-lzo ships its own copy of LZO, which keeps the packaged program
    # self contained (on Windows it is a single statically linked extension).
    import lzo as PYTHON_LZO
except ImportError:
    PYTHON_LZO = None

ctypes 那条路要按优先级找库文件，从最具体的地方开始：

def lzo_max_output_size(size):
    """Worst case size of a lzo1x_1 compressed block."""
    return size + size // 16 + 64 + 3

def _lzo_library_candidates():
    """Places to look for a liblzo compatible library, most specific first."""
    names = ("liblzo2.so.2", "liblzo2.so", "lzo2.dll", "minilzo.dll")
    directories = []
    bundle_dir = getattr(sys, "_MEIPASS", None)
    if bundle_dir:  # PyInstaller onefile unpack dir
        directories.append(bundle_dir)
    directories.append(os.path.dirname(os.path.abspath(__file__)))
    candidates = [os.getenv("MINILZO_PATH")]
    for directory in directories:
        candidates += [os.path.join(directory, name) for name in names]
    candidates += [
        ctypes.util.find_library("lzo2"),
        ctypes.util.find_library("minilzo"),
    ]
    return candidates

lzo_max_output_size 那个公式来自 LZO 文档的最坏情况上界，缓冲区按它开就不会溢出；sys._MEIPASS 是 PyInstaller 单文件运行时的解包目录——打包后的 DLL 就躺在那里

加载失败要静默跳过，候选列表里必然有一些不存在，抛错就没法接着往下找了：

def _load_lzo_library():
    for candidate in _lzo_library_candidates():
        if not candidate:
            continue
        try:
            return ctypes.cdll.LoadLibrary(candidate)
        except OSError:
            continue
    return None

真正的调用最容易写错，下面这个坑要到 64 位上才暴露：

def _compress_into(image_bytes, source, work_memory):
    # create_string_buffer() appends a NUL terminator, so the row length has to
    # be passed explicitly: the printer expects exactly one line.
    source_length = ctypes.c_size_t(len(image_bytes))
    destination = ctypes.create_string_buffer(lzo_max_output_size(len(image_bytes)))
    # lzo_uint is a size_t; a c_int would be written past on 64-bit platforms.
    destination_length = ctypes.c_size_t(len(destination))
    LZO.lzo1x_1_compress(
        source,
        source_length,
        destination,
        ctypes.byref(destination_length),
        work_memory,
    )
    return destination.raw[: destination_length.value]

这三个细节全都是踩出来的：

• create_string_buffer 会补一个 \0：所以必须显式传长度，否则打印机收到的是 81 字节（最后多一个 \0），整行错位
• lzo_uint 在 64 位上是 size_t，一开始我按 c_int 传，lzo1x_1_compress 会往那个 4 字节变量里写 8 字节，把相邻内存写穿，改成 c_size_t 才稳
• 工作内存要复用：lzo1x_1 需要一块 WORK_MEMORY_SIZE 大小的便签纸，每行开一块新的纯属浪费，所以全局只建一次

工作内存的管理：

LZO = None if PYTHON_LZO else _load_lzo_library()
_work_memory = None

def _compress_with_ctypes(image_bytes):
    global _work_memory
    if LZO is None:
        raise RuntimeError(
            "no LZO backend available: install python-lzo or point MINILZO_PATH "
            "at a liblzo2/minilzo shared library"
        )
    if _work_memory is None:
        _work_memory = ctypes.create_string_buffer(WORK_MEMORY_SIZE)
    source = ctypes.create_string_buffer(image_bytes)
    return _compress_into(image_bytes, source, _work_memory)

「两个后端都没有」时的报错信息里带了 MINILZO_PATH，这是给打包后的现场排查留的：用户可以把一个 minilzo.dll 放在旁边，用环境变量指过去

0x06 传输层：连不上时到底发生了什么

协议对了不代表连得上；蓝牙连接失败的花样很多：没配对、没开机、被手机占着、channel 不对；poooli_cli/transport.py 负责建 socket、重试

第一件事就是绕开 pybluez2 的一个坑：

def rfcomm_socket(bluetooth):
    """新建一个 RFCOMM 套接字。

    pybluez2 0.46 的 native_socket 拿 btcommon.Protocols 枚举成员当协议表的键，
    而 bluetooth.RFCOMM 只是整数常量 3，于是
    BluetoothSocket(bluetooth.RFCOMM) 会抛 KeyError: 3。优先传枚举成员，
    拿不到就退回构造函数默认值（同样是 RFCOMM）。
    """
    protocols = getattr(getattr(bluetooth, "btcommon", None), "Protocols", None)
    rfcomm = getattr(protocols, "RFCOMM", None)
    if rfcomm is not None:
        try:
            return bluetooth.BluetoothSocket(rfcomm)
        except KeyError:
            pass
    return bluetooth.BluetoothSocket()

这行代码让我卡了半天：文档里的写法 bluetooth.BluetoothSocket(bluetooth.RFCOMM) 在 0.46 上直接 KeyError: 3；原因是它的协议表用枚举成员当键，而 bluetooth.RFCOMM 是个整数；所以这里「优先枚举、连不上退回默认」

超时设置也是必须的：

def open_bluetooth_socket(bluetooth, args, interrupt):
    sock = rfcomm_socket(bluetooth)
    interrupt.watch(sock)
    if args.timeout > 0:
        sock.settimeout(args.timeout)
        LOG.debug(
            "socket 超时 %.1f 秒（阻塞调用会定期返回，Ctrl+C 才能及时生效）",
            args.timeout,
        )
    else:
        LOG.debug("socket 不设超时；真被系统调用卡住只能靠退出兜底定时器")
    return sock

为什么强调超时：不设超时的阻塞调用会把 Ctrl+C 一起吞掉；connect() 卡在内核里的时候，Python 要等它返回才能跑信号处理器；设了超时，调用定期返回，中断才有机会生效

重试循环里有一条很容易忽略的优先级：中断不算连接失败

def connect_printer(bluetooth, args, interrupt):
    """连上打印机（按 args.retries 重试），返回 (socket, Poooli)。"""
    attempts = max(1, args.retries)
    for attempt in range(1, attempts + 1):
        sock = open_bluetooth_socket(bluetooth, args, interrupt)
        printer = Poooli(sock)
        try:
            LOG.info("连接 %s (channel %d) ...", args.mac, args.channel)
            printer.connect(args.mac, args.channel)
            return sock, printer
        except OSError as error:
            close_quietly(sock)
            interrupt.watch(None)
            if interrupt.requested:
                raise KeyboardInterrupt  # 是 Ctrl+C / EOF 打断的，不算连接失败
            if attempt == attempts:
                raise  # 最后一次交给 commands 统一汇报
            LOG.error(
                "连接失败（第 %d/%d 次）：%s",
                attempt,
                attempts,
                describe_bluetooth_error(error),
            )
            if args.diagnose and attempt == 1:
                diagnose_connect(bluetooth, args.mac)  # 诊断一次就够，别拖慢重试
            LOG.info("等待 %.1f 秒后重试 ...", args.retry_delay)
            time.sleep(args.retry_delay)
    raise AssertionError("unreachable")  # pragma: no cover

if interrupt.requested: raise KeyboardInterrupt 这行看着多余，其实很关键：用户按了 Ctrl+C，connect 抛出来的也是一个 OSError（超时或中断）；如果当成「连接失败」去重试，用户按了三次 Ctrl+C 都退不出来

诊断模块把 Winsock 错误码翻译成人话：

WINERROR_HINTS = {
    10038: "socket 已经关了（多半是刚收到 Ctrl+C / EOF，正在退出）",
    10048: "地址已被占用：上一次连接可能没断干净，等几秒或重启打印机",
    10049: "地址不可用：检查 -m 里的 MAC 地址有没有写错",
    10050: "蓝牙适配器不可用：先确认 Windows 的蓝牙是打开的",
    10051: "Windows 到不了这个设备。常见原因：打印机没开机/不在范围、还没在系统里配对、"
    "或者正被手机 App 连着（啵哩一次只接受一个连接）",
    10060: "连接超时：设备在范围内但没应答，多半是没配对，或者 channel 不对",
    10061: "连接被拒绝：打印机可能正被别的设备连着，或者 channel 不对（换 -c 2 / -c 3 试试）",
    10064: "目标设备已关闭或暂时不可达",
    10065: "主机不可达：打印机没开机、不在范围或已休眠",
}

CONNECT_CHECKLIST = (
    "打印机开机、亮灯；手机 App / 其他电脑先断开（啵哩一次只接一个连接）",
    "在 Windows「设置 → 蓝牙和其他设备」里把打印机配对好，未配对时 RFCOMM 连不上",
    "poooli.exe --list 里要能看到这个 MAC；扫不到就是没开机 / 不在范围 / 地址不对",
    "换 channel 试试（-c 2 / -c 3）；加 -v 看完整日志",
)

def describe_bluetooth_error(error):
    """把 Winsock 错误码翻成能照着做的提示。"""
    message = "%s" % error
    hint = WINERROR_HINTS.get(getattr(error, "winerror", None))
    return "%s（%s）" % (message, hint) if hint else message

连接失败一次之后，还会做一次现场排查，用来区分「扫不到设备」（没开机 / 没配对）、「扫得到但没有 SPP 服务」（多半是 BLE 设备）、「有服务却连不上」（被占用 / channel 不对）：

def diagnose_connect(bluetooth, mac):
    """连接失败后的现场诊断；这里的异常都自己吞掉，免得盖住原始错误。"""
    LOG.info("开始诊断 %s ...", mac)
    try:
        name = bluetooth.lookup_name(mac, timeout=5)
    except Exception as error:  # noqa: BLE001 - 后端异常五花八门
        LOG.warning("名称查询失败：%s", error)
    else:
        LOG.info(
            "名称查询：%s", name or "没应答（没开机 / 不在范围 / 已被别的设备占用）"
        )

    try:
        devices = bluetooth.discover_devices(duration=4, lookup_names=True)
    except Exception as error:  # noqa: BLE001
        LOG.warning("设备扫描失败：%s", error)
    else:
        found = [device for device in devices if device[0].upper() == mac.upper()]
        LOG.info(
            "扫描到 %d 个设备；%s",
            len(devices),
            "其中包含目标 %s" % (found[0],)
            if found
            else "没有目标 %s（已配对的设备不一定会出现在扫描结果里）" % mac,
        )

    try:
        services = bluetooth.find_service(address=mac)
    except Exception as error:  # noqa: BLE001
        LOG.warning("SDP 服务查询失败：%s", error)
    else:
        if not services:
            LOG.warning(
                "SDP 没查到服务：设备没配对 / 不在范围，或者根本不是经典蓝牙"
                "（BLE 设备不能用 RFCOMM）"
            )
        for service in services:
            LOG.info(
                "SDP 服务：name=%s protocol=%s port=%s classes=%s",
                service.get("name"),
                service.get("protocol"),
                service.get("port"),
                service.get("service-classes"),
            )
    LOG.info("诊断结束")

0x07 文本：把一串字排成纸

文本是这个项目第二个大功能；它要解决的是排版，画字只是其中一步：中文没有空格怎么断行、标点不能掉行首、找不到中文字体怎么办、内容太长怎么分页

拆成四个模块：fonts 找字体、layout 折行分页、render 画位图、dump 存图调试

找字体

字体不能写死；Windows 上有微软雅黑，Linux 上多半是 Noto CJK，macOS 上是 PingFang；所以每个平台排一份候选表：

CANDIDATE_FONTS = {
    "win32": (
        "msyh.ttc",  # 微软雅黑，带中文
        "msyhbd.ttc",
        "simhei.ttf",  # 黑体
        "simsun.ttc",  # 宋体
        "Deng.ttf",  # 等线
        "consola.ttf",  # 等宽
        "segoeui.ttf",
        "arial.ttf",
        "cour.ttf",
    ),
    "darwin": (
        "/System/Library/Fonts/PingFang.ttc",
        "/System/Library/Fonts/STHeiti Light.ttc",
        "/System/Library/Fonts/Supplemental/Songti.ttc",
        "/Library/Fonts/Arial.ttf",
        "/System/Library/Fonts/Menlo.ttc",
    ),
    "linux": (
        "NotoSansCJK-Regular.ttc",
        "NotoSansCJKsc-Regular.otf",
        "wqy-zenhei.ttc",
        "wqy-microhei.ttc",
        "SourceHanSansSC-Regular.otf",
        "DejaVuSansMono.ttf",
        "DejaVuSans.ttf",
    ),
}

搜索是按优先级来的，没有用「找到哪个算哪个」：_search 拿文件名当下标排名，找到更靠前的就替换，rank 0 直接提前返回

def _search(directory, names):
    """在目录里递归找候选字体，按 names 的顺序优先，找到第一个就返回。"""
    wanted = {name.lower(): rank for rank, name in enumerate(names)}
    best = None
    for root, _dirs, files in os.walk(directory):
        for name in files:
            rank = wanted.get(name.lower())
            if rank is None:
                continue
            if best is None or rank < best[0]:
                best = (rank, Path(root, name))
                if rank == 0:
                    return best[1]
    return best[1] if best is not None else None

加载失败绝不静默：

def load_font(path=None, size=32):
    """加载字体。找不到系统字体时退回 Pillow 内置字体并告警。"""
    if path:
        if not Path(path).is_file():
            raise FileNotFoundError("找不到字体文件：%s" % path)
        font = ImageFont.truetype(str(path), size)
        LOG.debug("使用指定字体：%s", describe_font(font))
        return font

    found = find_font_file()
    if found is not None:
        font = ImageFont.truetype(str(found), size)
        LOG.debug("自动选中字体：%s", describe_font(font))
        return font

    LOG.warning(
        "没找到系统字体，退回 Pillow 内置字体：中文会渲染成方块，"
        "请用 --font 指定一个 .ttf/.ttc（Windows 上例如 C:\\Windows\\Fonts\\msyh.ttc）"
    )
    try:
        return ImageFont.load_default(size=size)
    except TypeError:  # 老 Pillow 的 load_default() 不收 size
        return ImageFont.load_default()

这条 WARNING 很重要；Pillow 内置字体只有拉丁字符，中文会渲染成方块，但程序不会报错——用户只会看到一张印满方块的纸，然后来问为什么；所以宁可吵一点，也要在日志里说清楚

折行：中文禁则

折行的核心矛盾：中文没有空格，不能按词断行，只能按字断；但按字断会出问题：标点跑到行首（。 单独占一行）非常难看；所以要做「禁则」

先把文本切成断行单元；规则是：中文单字一个单元，拉丁词一个单元，空格跟着前一个：

def is_wide(char):
    """东亚全角字符（可以逐字断行）。"""
    return unicodedata.east_asian_width(char) in ("W", "F")

def tokens(paragraph):
    """把一段话切成断行单元：中文单字一个单元，拉丁词一个单元，空格跟着前一个。"""
    result = []
    for char in paragraph:
        if char == " ":
            if result and not result[-1].endswith(" "):
                result[-1] += " "
            else:
                result.append(" ")
        elif (
            result
            and result[-1][-1] != " "
            and not is_wide(char)
            and not is_wide(result[-1][-1])
        ):
            result[-1] += char
        else:
            result.append(char)
    return result

判定用的是 east_asian_width，不是「是不是中文」——日文、韩文、全角标点都会落在 W/F 里，一并处理掉了

然后是折行主体，禁则就体现在两个 continue 上：

# 不能出现在行首的标点（简单版禁则）。
NO_LINE_START = "，。、；：？！）】》」』%,.;:?!)]}>"
NO_LINE_END = "（【《「『([{<"

def wrap_paragraph(paragraph, font, max_width):
    """把一段（不含换行的）文本折成若干行。"""
    lines = []
    current = ""
    for token in tokens(paragraph):
        if token and font.getlength(token) > max_width:
            if current:
                lines.append(current.rstrip())
            pieces = _hard_split(token, font, max_width)
            lines.extend(pieces[:-1])
            current = pieces[-1]
            continue
        if token in NO_LINE_START and current:
            current += token  # 标点不给放行首
            continue
        candidate = current + token
        if current and font.getlength(candidate.rstrip()) > max_width:
            if current[-1] in NO_LINE_END:  # 左括号不给放行尾
                current += token
                continue
            lines.append(current.rstrip())
            current = token.lstrip(" ") if token.startswith(" ") else token
        else:
            current = candidate
    lines.append(current.rstrip())
    return [line for line in lines if line != ""] or [""]

注意测量用的是 font.getlength()，不是「字数 × 字号」；等宽拉丁字符和全角汉字宽度差一倍，按字数算必然错位

超长单元（比如一条没有空格的 URL）要硬切：

def _hard_split(token, font, max_width):
    """把超过一整行的单元按字符切成能放下的碎片。"""
    pieces = []
    current = ""
    for char in token:
        if current and font.getlength(current + char) > max_width:
            pieces.append(current)
            current = ""
        current += char
    if current:
        pieces.append(current)
    return pieces

再往上一层是「整段折行」和「分页」：

def wrap(text, font, max_width):
    """整段文本折行：显式换行保留，空行也保留。"""
    lines = []
    for paragraph in expand_tabs(unescape(text)).split("\n"):
        if not paragraph.strip():
            lines.append("")
            continue
        lines.extend(wrap_paragraph(paragraph, font, max_width))
    return lines

@dataclass
class Page:
    """一页：渲染用的行 + 这一页在原文里的行号范围。"""
    lines: list
    first_line: int

    @property
    def count(self):
        return len(self.lines)

def paginate(lines, lines_per_page):
    """按每页行数切页；lines_per_page <= 0 表示不分页。"""
    if lines_per_page is None or lines_per_page <= 0:
        return [Page(lines=list(lines), first_line=0)]
    pages = []
    for index in range(0, len(lines), lines_per_page):
        pages.append(
            Page(lines=list(lines[index : index + lines_per_page]), first_line=index)
        )
    return pages or [Page(lines=[], first_line=0)]

def lines_per_page_for(page_height, line_height):
    """把像素页高换算成行数；<=0 表示不分页。"""
    if page_height is None or page_height <= 0:
        return 0
    return max(1, page_height // line_height)

Page 带 first_line 是为了日志能说「第 3 页从原文第 41 行开始」，分页出了问题能直接定位

命令行里的 \n 转义也在这里处理：

def unescape(text):
    r"""把命令行里写的字面 \n \t \\ 变成真实字符（只处理这三个）。"""
    out = []
    index = 0
    mapping = {"n": "\n", "t": "\t", "\\": "\\", "r": "\r"}
    while index < len(text):
        char = text[index]
        if char == "\\" and index + 1 < len(text) and text[index + 1] in mapping:
            out.append(mapping[text[index + 1]])
            index += 2
            continue
        out.append(char)
        index += 1
    return "".join(out)

只处理这四个转义，不做通用解析——--text 里的反斜杠大多是路径（C:\Users\...），过度解析反而会吃掉用户的字符

渲染：行高、对齐、不留白

DEFAULT_WIDTH = poooli.Poooli.WIDTH  # 640px，和协议里写死的纸宽一致
DEFAULT_MARGIN = 16
DEFAULT_LINE_SPACING = 1.35
ALIGNMENTS = ("left", "center", "right")

def line_height(font, line_spacing=DEFAULT_LINE_SPACING):
    """一行占多少像素（含行距）。"""
    ascent, descent = font.getmetrics()
    return max(1, int(round((ascent + descent) * line_spacing)))

行高用 ascent + descent，不用 font.size；有些字体的实际高度会超过字号，按字号算会让上下行贴在一起

render_page 有一个细节：末尾的空行不参与高度计算，免得浪费纸

def render_page(
    lines,
    font,
    width=DEFAULT_WIDTH,
    margin=DEFAULT_MARGIN,
    align="left",
    line_spacing=DEFAULT_LINE_SPACING,
):
    """把一页的行渲染成位图；末尾的空行不会浪费纸。"""
    if align not in ALIGNMENTS:
        raise ValueError("Unknown align: " + align)
    step = line_height(font, line_spacing)
    used = 0
    for index, line in enumerate(lines):
        if line.strip():
            used = index + 1
    height = max(step, 2 * margin + used * step)
    image = Image.new("RGB", (width, height), "white")
    draw = ImageDraw.Draw(image)
    usable = max(1, width - 2 * margin)
    for index, line in enumerate(lines):
        if not line:
            continue
        offset = draw.textlength(line, font=font)
        if align == "center":
            x = margin + (usable - offset) / 2
        elif align == "right":
            x = margin + usable - offset
        else:
            x = margin
        draw.text((x, margin + index * step), line, font=font, fill="black")
    LOG.debug(
        "渲染一页：%dx%d，%d 行（%s 对齐）", image.width, image.height, used, align
    )
    return image

这一步画出来的是白底黑字的 RGB 图，不是二值图；为什么？因为 PIL 画字的抗锯齿只有灰度才有意义；真正的二值化留到 prepare_image，和图片走同一条路；这样「文本」和「图片」在协议层完全没有区别——都是 640 宽的二值位图

最外层的 render_pages 把四步串起来：

def render_pages(
    text,
    font_path=None,
    font_size=32,
    width=DEFAULT_WIDTH,
    margin=DEFAULT_MARGIN,
    align="left",
    line_spacing=DEFAULT_LINE_SPACING,
    page_height=0,
    lines_per_page=0,
):
    """排版并渲染整段文本，返回每页一张 PIL 位图（640px 宽）。"""
    font = load_font(font_path, font_size)
    max_width = max(1, width - 2 * margin)
    lines = wrap(text, font, max_width)
    step = line_height(font, line_spacing)
    per_page = lines_per_page or lines_per_page_for(page_height, step)
    pages = paginate(lines, per_page)
    LOG.info(
        "文本排版：%d 行 → %d 页（%s，每页最多 %s 行，行高 %dpx）",
        len(lines),
        len(pages),
        describe_font(font),
        per_page or "不限",
        step,
    )
    for page in pages:
        LOG.debug(
            "第 %d 页首行（原文第 %d 行）：%s",
            pages.index(page) + 1,
            page.first_line + 1,
            (page.lines[0] if page.lines else "")[:40],
        )
    return [
        render_page(page.lines, font, width, margin, align, line_spacing)
        for page in pages
    ]

存图调试：--verbose 时把「打印机收到什么」留下来

文字打印最烦的问题是「印出来不对」——字体错了、折行错了、二值化阈值不合适；光看纸很难判断，所以加了「把渲染出的位图存成 PNG」：

def save_pages(pages, directory=DEFAULT_DIR, prefix="page", start=1):
    """把每页存成 ``<prefix>-0001.png``，返回保存的路径列表。"""
    target = Path(directory)
    target.mkdir(parents=True, exist_ok=True)
    saved = []
    for index, page in enumerate(pages, start):
        path = target / ("%s-%04d.png" % (prefix, index))
        page.save(path)
        saved.append(path)
    return saved

def dump_directory(selected, verbose=False, default=DEFAULT_DIR):
    """算出要不要存图、存到哪。

    `selected` 是选项给的值：``None`` 表示没给（`--verbose` 时用默认目录），
    ``False`` 表示 `--no-dump-images`（死活不存），其余当目录用。
    """
    if selected is None:
        return default if verbose else None
    return selected or None

--no-dump-images 用 store_const 存成 False，和「没给」（None）区分开——这样 --no-dump-images 的优先级高于 -v，用户表达「我不要这些文件」时不会被覆盖

0x08 ssh：把终端会话镜像到纸上

这是项目里最大胆的一块功能：poooli ssh --mac AA:BB:... root@host，你在电脑上看到什么，打印机上就有什么

它要同时解决四件事，每一件单独看都不难，叠在一起就很烦：

1. 跑一个真正的 ssh 子进程（不是自己实现 ssh），参数原样转发；
2. POSIX 上给它 pty，本地终端切 raw，按键原样转发；
3. 子进程的 stdout/stderr 合成一路：一边显示到终端（保留颜色），一边剥掉 ANSI 转义送打印机；
4. 节流：打印机的速度远低于终端，不能让每个回车都变成一张纸

剥 ANSI：打印机不认颜色

终端输出里什么都有：颜色、光标移动、进度条的回车、清屏；先剥掉转义序列，只留看得见的字符；但 \n、\r、\b 要留给下一层，因为它们有语义：

# CSI（含颜色）/ OSC（窗口标题等）/ 其它两字符转义
ESCAPE = re.compile(
    r"""
    \x1b
    (?:
        \[ [0-?]* [ -/]* [@-~]          # CSI ...
      | \] [^\x07\x1b]* (?:\x07|\x1b\\)  # OSC ... BEL 或 ST
      | [@-Z\\-_]                        # Fe 两字符序列
    )
""",
    re.VERBOSE,
)

# 其它 C0 控制符（保留 \t \n \r \b，交给 LineBuffer）
CONTROL = re.compile(r"[\x00-\x07\x0b\x0c\x0e-\x1f\x7f]")

def strip(text):
    """去掉转义序列与不可见控制符；\t \n \r \b 原样保留。"""
    return CONTROL.sub("", ESCAPE.sub("", text))

LineBuffer：\r 回到行首覆盖

这是整个 ssh 功能里最容易写错的地方；终端的换行是 \r\n：回车 + 换行，进度条又靠 \r 原地刷新；如果看到 \r 就把当前行清空，那 abc\r\n 里的 abc 会被一起丢掉

正确做法是维护一个光标位置，\r 把光标移回 0，后面的字符覆盖上去：

class LineBuffer:
    """按行切分终端文本：``\n`` 收一行，``\r`` 回行首覆盖，``\b`` 退格删一个字。

    关键在 ``\r``：pty 上每行以 ``\r\n`` 结尾，进度条又靠 ``\r`` 原地刷新，所以不能
    简单地把当前行清空（那会把 ``abc\r\n`` 里的 ``abc`` 一起丢掉），要按光标位置
    覆盖。``carriage_return_newline=True`` 换一种语义：``\r`` 也算换行（raw 终端里
    回车键送出的是 ``\r``），用来处理键盘输入。
    """

    def __init__(self, carriage_return_newline=False):
        self.carriage_return_newline = carriage_return_newline
        self.chars = []
        self.cursor = 0
        self._just_newline = False

    @property
    def text(self):
        return "".join(self.chars)

    def reset(self):
        self.chars = []
        self.cursor = 0

    def feed(self, text):
        """吃一段文本，返回其中已经完整的行。"""
        if self._clean(text):
            if "\n" not in text:
                self.chars.extend(text)
                self.cursor = len(self.chars)
                return []
            *lines, tail = (self.text + text).split("\n")
            self.reset()
            self.chars.extend(tail)
            self.cursor = len(self.chars)
            return lines
        lines = []
        for char in text:
            if char == "\n":
                if not self._just_newline:
                    lines.append(self.text)
                self.reset()
                self._just_newline = False
            elif char == "\r":
                if self.carriage_return_newline:
                    lines.append(self.text)  # raw 终端的回车键就是换行
                    self.reset()
                    self._just_newline = True
                else:
                    self.cursor = 0  # 回到行首，后面的字覆盖上来
            elif char == "\b":
                if self.cursor:
                    self.cursor -= 1
                    del self.chars[self.cursor]
            elif self.cursor < len(self.chars):
                self.chars[self.cursor] = char
                self.cursor += 1
            else:
                self.chars.append(char)
                self.cursor += 1
        return lines

    def _clean(self, text):
        """这段文本能不能走快路径（没有光标控制，且当前光标就在行尾）。"""
        return (
            self.cursor == len(self.chars)
            and not self._just_newline
            and "\r" not in text
            and "\b" not in text
        )

    def take(self):
        """取走还没发出去的半行（提示符、密码提示之类）。"""
        pending = self.text
        self.reset()
        self._just_newline = False
        return pending

_clean 是个优化：绝大多数输出块里没有 \r / \b，光标也在行尾，这时候可以直接 split("\n") 批量处理，不用逐字符跑状态机；终端流量很大，这个快路径是必要的

take() 处理的是「半行」：提示符 $ 后面没有换行，如果只按行发，提示符永远印不出来；所以空闲一段时间后要把残留的半行也发出去

节流：FLUSH_INTERVAL

打印机很慢；逐行发的话，跑一个 apt update 能印掉半卷纸；所以攒一段再发：

# 攒这么久再渲染一次（秒）；这是唯一的出处，其他地方只提常量名，改值只改这里。
# 这段时间里的行会合成同一张位图（0 表示来一行发一行）。
FLUSH_INTERVAL = 0.8

节流循环由 printer.tick() 驱动，主循环每 200ms 叫它一次：

def tick(self):
    """开一次缓冲区（pump 循环每 200ms 叫一次）。

    攒够 `flush_interval` 并且有新行才渲染；空闲够久再把提示符那样的半行
    也发出去。没有新内容的时候什么都不做——不渲染，也不发送。
    """
    interval = self.options.flush_interval
    if not interval or time.monotonic() - self.last_flush >= interval:
        self.flush_lines()
    idle = self.options.idle_flush
    if idle and time.monotonic() - self.last_activity >= idle:
        self.flush_pending()

def flush_lines(self):
    """把攒下的整行一次渲染出去；没有新行就返回 False。"""
    lines = self.take_lines()
    if not lines:
        return False
    self.last_flush = time.monotonic()
    self.printer.send_block(lines)
    return True

def flush_pending(self):
    """收尾：整行和没有换行的半行都立刻发出去（会话结束 / banner 前后）。"""
    lines = self.take_lines()
    for buffer in (self.output, self.stdin):
        pending = buffer.take()
        if pending and not is_console_noise(pending):
            lines.append(pending)
    self.last_flush = time.monotonic()
    self.printer.send_block(lines)

send_block 是省纸的关键：把这一批行合成一张位图发一次，逐行发的老路就省掉了；渲染省了，蓝牙往返也省了

def send(self, line):
    """发一行（等价于只有一行的块）。"""
    return self.send_block([line])

def send_block(self, lines):
    """把一批行一次渲染、一次发出：省渲染，也省一遍蓝牙往返。

    纯空行（包括只有空白的行）直接丢掉；整批都是空行就什么都不做。
    """
    texts = []
    for line in lines:
        text = line.rstrip()
        if not text:
            continue
        if self.timestamp:
            text = "%s %s" % (time.strftime("%H:%M:%S"), text)
        texts.append(text)
    if not texts:
        return False
    if self.printer is None and not self.dump_dir:
        LOG.debug("（只在本机显示）%s", "\n".join(texts))
        return False
    text = "\n".join(texts)
    try:
        pages = poooli_text.render_pages(
            text,
            font_path=self.font_path,
            font_size=self.font_size,
            margin=0,
            line_spacing=1.0,
        )
        self.dump(pages, text)
    except (OSError, ValueError) as error:
        LOG.error("这批行渲染失败，跳过：%s（%r）", error, text[:40])
        return False
    if self.printer is None:  # --no-print：只存图
        return False
    try:
        for page in pages:
            self.printer.send_image_object(page, self.mode)
    except (OSError, ValueError) as error:
        LOG.error("打印机出错，接下来只在本机显示：%s", error)
        self.printer = None
        self.failed = True
        return False
    self.lines_sent += len(texts)
    return True

注意 except 那段：打印机中途出错（没纸、走远、断电）时，会话本身要继续——用户还在 ssh 里干活，不能因为打印机出错就把 ssh 也断了；所以出错后把 self.printer = None，之后只在本机显示

滤掉 Windows ssh 的噪声

Windows 版 ssh（Win32-OpenSSH）在管道模式下会把「句柄不是控制台」的告警打到 stderr；句柄本来就不是控制台，这行纯属噪声，不该上纸：

# Windows 版 ssh（Win32-OpenSSH）在管道模式下会把「句柄不是控制台」的告警打到 stderr，
# 句柄本来就不是控制台，这行纯属噪声。首行匹配（不匹配行中间的相似文本）；整行的那种
# 连换行一起删掉，半行/跨块的残渣由 `is_console_noise` 在成行后再拦一道。
CONSOLE_NOISE = re.compile(
    r"(?m)^(?:Get|Set)ConsoleMode on [^\r\n]*failed with \d+[ \t]*\r?\n"
)
NOISE_LINE = re.compile(r"^(?:Get|Set)ConsoleMode on .*failed with \d+$")

def drop_console_noise(text):
    """删掉完整的 ssh 控制台告警行（半行留到拼成整行时再判）。"""
    return CONSOLE_NOISE.sub("", text)

def is_console_noise(line):
    """这一行是不是 ssh 的控制台告警。"""
    return bool(NOISE_LINE.match(line.strip()))

两道过滤是因为噪声可能跨块：告警的前半截在这个 4096 字节的读块里，后半截在下一个块里；所以块级删一次，成行后再判一次

本地显示与输入镜像

终端输出要同时满足两个消费者；本地这份要补 \r\n，因为 raw 模式下终端不会自己把 \n 变成回车换行（否则输出会「爬楼梯」）：

def write_output(self, text):
    text = drop_console_noise(text)  # 跨块被切断的那种在下面成行后再拦
    self.out.write(self._for_local_terminal(text))
    self.out.flush()
    self.last_activity = time.monotonic()
    self.enqueue(self.output.feed(strip(text)))

def _for_local_terminal(self, text):
    """本机终端切了 raw 就不再自己把 \\n 变成 \\r\\n，这里补上，否则输出会爬楼梯。"""
    if not self.raw_local:
        return text
    return text.replace("\r\n", "\n").replace("\n", "\r\n")

键盘输入要不要打到纸上是个判断题；远端有 pty 时会自己回显输入，再镜像一遍就重复了；管道模式下远端不回显，不打就看不到输入行；所以默认 auto，按「远端会不会回显」推断：

@property
def mirror_stdin(self):
    """远端有 pty 时它会自己回显，再镜像一遍就会重复。"""
    mode = self.options.stdin_mirror
    if mode == "always":
        return True
    if mode == "never":
        return False
    return not self.remote_echo

def write_input(self, text):
    if not self.mirror_stdin:
        return
    self.enqueue(
        ["> " + line for line in self.stdin.feed(strip(text)) if line.strip()]
    )

输入行加 > 前缀，纸上就能和命令输出区分开

跑真正的 ssh：pty、双向转发

会话层给子进程一个真正的 pty（和 openssh 一样），把 stdin 接过去，把 stdout/stderr 合成一路：

try:  # POSIX only
    import fcntl
    import pty
    import struct
    import termios
    import tty
except ImportError:  # pragma: no cover - Windows
    fcntl = pty = struct = termios = tty = None

def default_pty():
    """有 pty 模块就用；Windows 上退回管道。"""
    return pty is not None and sys.stdin is not None and hasattr(sys.stdin, "fileno")

Windows 上没有 pty 模块，退回管道，并自动给 ssh 补 -tt 让远端分配伪终端：

def remote_pty_wanted(ssh_args):
    """ssh 会不会给远端分配伪终端（有伪终端远端就会自己回显输入）。

    本地有 pty 时 ssh 默认分配；没有本地 pty 时 `command.build_argv` 会补 `-tt`。
    """

主循环要处理一个容易被忽略的收尾问题：子进程退出后，管道里可能还有输出没读完；如果立刻返回，最后几行就丢了：

def pump(self, mirror, on_tick=None):
    """转发到子进程退出；返回子进程退出码。

    子进程退出前可能还有一大段输出没读完（管道是边写边读的），所以退出后要
    等 reader 把剩下的读完、打完再收工，否则最后几行会丢。
    """
    reader = threading.Thread(
        target=self._reader, args=(mirror,), name="ssh-out", daemon=True
    )
    reader.start()
    threading.Thread(
        target=self._writer,
        args=(mirror, not self.use_pty),
        name="ssh-in",
        daemon=True,
    ).start()
    while True:
        try:
            code = self.child.wait(timeout=POLL_INTERVAL)
            break
        except subprocess.TimeoutExpired:
            if self._stopped.is_set():
                self.stop()
                try:
                    code = self.child.wait(timeout=5)
                except subprocess.TimeoutExpired:  # pragma: no cover
                    self.child.kill()
                    code = self.child.wait()
                break
            if on_tick is not None:
                on_tick()
    reader.join(DRAIN_TIMEOUT)
    if reader.is_alive():  # pragma: no cover
        LOG.debug("还有输出没读完（打印机太慢或管道没关），先收工")
    flush = getattr(mirror, "flush_pending", None)
    if flush is not None:
        flush()  # 缓冲区里攒的整行、还有没换行的半行，收工前一起发出去
    return code

终端窗口大小也要同步给远端，否则 vim、top 这类全屏程序会画错：

def set_window_size(fd, rows, columns):
    if termios is None:  # pragma: no cover - Windows
        return
    try:
        fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, columns, 0, 0))
    except OSError as error:  # pragma: no cover
        LOG.debug("设置终端大小失败：%s", error)

def install_resize_handler(session):
    """终端窗口变化时同步给远端；返回原来的处理器，会话结束后还原。"""
    import signal

    def handler(signum, frame):  # pragma: no cover
        rows, columns = terminal_size()
        session.resize(rows, columns)

    if not hasattr(signal, "SIGWINCH"):
        return None
    try:
        previous = signal.getsignal(signal.SIGWINCH)
        signal.signal(signal.SIGWINCH, handler)
    except (OSError, ValueError) as error:  # pragma: no cover
        LOG.debug("接管 SIGWINCH 失败：%s", error)
        return None
    return previous

会话的总装放在最后；finally 里做了三件收尾：还原 SIGWINCH、关闭子进程、把缓冲区里的内容刷出去（包括 footer）：

def run_session(ssh_argv, printer=None, options=None, interrupt=None):
    """跑一次 ssh 会话，顺带把内容打到打印机；返回 SessionResult。"""
    options = options or SessionOptions()
    use_pty = default_pty() if options.use_pty is None else options.use_pty
    if use_pty != default_pty():  # 用户显式指定
        LOG.debug("pty 模式：%s", use_pty)
    if options.remote_echo is None:
        options.remote_echo = remote_pty_wanted(ssh_argv[1:])
    if options.stdin_mirror == "auto" and not options.remote_echo:
        LOG.debug("远端不会回显输入，所以输入行也打到纸上")

    mirror = SessionMirror(printer, options=options)
    if options.banner:
        mirror.note(options.banner)

    session = SshSession(ssh_argv, use_pty=use_pty)
    started = time.monotonic()
    interrupted = False
    code = 0
    previous_handler = None
    try:
        session.start()
        previous_handler = install_resize_handler(session)
        if interrupt is not None:
            interrupt.watch(_InterruptSocket(session))
        code = session.pump(mirror, on_tick=mirror.tick)
    except KeyboardInterrupt:
        LOG.warning(
            "已中断（%s）", getattr(interrupt, "reason", None) or "KeyboardInterrupt"
        )
        interrupted = True
        session.stop()
    finally:
        restore_resize_handler(previous_handler)
        session.close()
        mirror.flush_pending()
        if options.footer:
            mirror.note(options.footer)

    result = SessionResult(
        code=code or 0, interrupted=interrupted, lines_sent=mirror.printer.lines_sent
    )
    LOG.info(
        "ssh 会话结束：退出码 %s，用时 %.1f 秒，打印机收到 %d 行",
        result.code,
        time.monotonic() - started,
        result.lines_sent,
    )
    return result

自己的选项怎么和 ssh 的选项共存

poooli ssh 的目标是「用法和 openssh-client 几乎一致」；难点在于参数归属：-p、-i、-L 是 ssh 的，--mac、--font-size 是我的

规则定得很死：自己只认长选项，短选项全部留给 ssh；所以 -v 是 ssh 的 verbose，--verbose 才是本程序的日志开关；-m 是 ssh 的 MAC 算法，打印机地址只能用 --mac

class OptionError(ValueError):
    """我们自己的长选项写错了（缺值、类型不对、取值不在候选里）。"""

@dataclass(frozen=True)
class Option:
    """一个只属于我们的长选项。"""
    name: str
    dest: str
    kind: str  # "value"（要跟一个值）或 "flag"（开关）
    default: object
    help: str
    metavar: str = "VALUE"
    choices: tuple = ()
    const: object = True  # flag 的取值，--no-xxx 那种用 False
    convert: object = None  # value 的转换函数（int / float）

def _value(name, dest, default, help, **kwargs):
    return Option(name, dest, "value", default, help, **kwargs)

def _flag(name, dest, default, help, const=True):
    return Option(name, dest, "flag", default, help, const=const)