简介ASTM E1381/1394协议是医疗检验设备与信息系统之间数据交换的通用标准广泛应用于检验仪器与LIS实验室信息系统通信。这一Python实现面向医疗信息化工程师、系统集成商及中间件开发者提供完整的ASTM数据解码与编码能力既能将接收到的ASTM消息解析为Python对象也能将业务数据编码回标准报文并基于此构建客户端与服务器端通信程序。压缩包共42个文件大小仅66KB体积十分轻量其中27个Python源文件构成核心功能按协议层、记录层、映射层等模块划分涵盖编解码器、异常处理、兼容性工具等另有10个RST文档用于接口说明和协议参考便于二次开发与学习。目前已有184人学习下载适合具备Python基础、需要快速接入医疗仪器或开发分析仪驱动的开发者。借助该资源可以直接复用成熟的ASTM消息处理链路省去逐字节解析协议的重复劳动同时内置的常见规范例程与现成驱动思路为开发自己的中间件、分析仪接入方案或软硬件网关提供了清晰范本。1. 为什么还要用Python实现ASTM E1381/1394数据协议ASTM E1381与E1394并不是同一个标准的新旧版本而是配对出现的两层协议规范。E1381解决的是低级传输问题比如怎样把一个消息切成帧帧序号怎么循环校验和怎么算E1394解决的是内容问题比如H、P、O、R、C、L这些记录里字段如何摆放。很多医院检验科里的血球仪、生化仪、免疫分析仪至今仍保留着RS-232接口并默认输出这套协议。接入这类设备时并不一定要依赖厂商专用转发软件用Python在中间层直接完成帧解析、记录转换和结果入库反而更容易控制和扩展。下面会一步步把帧层、消息层和串口收发写出来并提供可运行的测试用例。提示题目标题把两个标准写成E1381/1394表示在现场它们是打包出现的以下实现也按“帧层消息层”的组合展开。2. ASTM E1381/1394协议核心帧格式、校验和与状态机2.1 帧格式控制字符与两层协议想象你从仪器的串口读到一段这样的字节流02 32 02 48 7C 5C 5E 26 7C ... 03 34 31 0D 0A也就是STX2STXH|\^|...ETX41CRLF。第一个STX表示一帧开始随后是帧序号单字符0到7每发送一帧循环加1然后又是STX随后是文本即E1394的记录内容到ETX结束再跟两个十六进制字符作为校验和最后是CR LF。ASTM E1381只关心这个外壳E1394则规定文本里的记录。这样分层的好处是帧层可以复用记录层的字段变更不会动摇底层收发。在真正写代码前先把控制字符的作用整理成一张表。控制字符ASCII码本协议中的角色STX0x02帧开始也作为帧序号与记录文本之间的分隔ETX0x03记录文本结束校验和从这里之后开始CR0x0D帧结束同时也常作为记录分隔符LF0x0A帧结束的跟随符有的设备省略ENQ0x05请求建立传输相当于“我要开始发了”ACK0x06对ENQ或一帧数据的确认NAK0x15否定应答常见原因是校验和错误EOT0x04本次传输会话结束帧序号不能从帧文本中读出来而要在状态机里识别。文本部分并不包含帧序号也不包含第二个STX这些信息都在帧层里。这也是为什么解析时不能拿字符串split()一带而过必须按字节位置处理。2.2 校验和算法两个十六进制字符怎么算出来的E1381的校验和字段是两位ASCII十六进制字符表示帧内容的二进制和。具体求和范围各厂商代码里并不完全一致最常见做法是“从帧序号起到ETX为止所有字节求和后取低8位”。少数仪器会把第一个STX也纳入实现时把求和范围的起点作为可配置参数能省去很多现场麻烦。def compute_checksum(body: bytes) - str: 对帧正文求和结果转成两位大写十六进制。 total 0 for b in body: total (total b) % 256 return f{total:02X}以一个具体帧为例帧序号是2内层也是STX文本为H|...那么求和对象是32 02 48 7C 03。把它们的ASCII码值相加取低八位再格式化成41这样的字符串。如果结果是0A之类带字母的数值直接按ASCII字符放在帧里不需要转换成控制字符。这个函数是后面所有验证的基础。实际接入仪器时建议先用设备自带的一条样本报文把仪器手册上给的示例帧手工加一遍确认厂商是不是把STX纳入计算。下面构造帧的代码会提供一个include_stx_in_checksum参数默认按最常见的范围算。2.3 接收状态机从字节流把完整帧切出来串口是字节流没有现成的“一条消息”概念。所以接收端要维护一个状态机按顺序经历IDLE、BODY、CKSUM、TAIL阶段直到CR/LF出现才算凑齐一帧。实现里把每个字节喂给状态机状态机在帧结束时把原始帧交出来。class FrameReceiver: def __init__(self): self.reset() def reset(self): self.buffer bytearray() self.cks bytearray() self.stage IDLE def feed(self, b: int): if self.stage IDLE: if b 0x02: self.buffer.clear() self.stage BODY elif self.stage BODY: if b 0x03: self.stage CKSUM else: self.buffer.append(b) elif self.stage CKSUM: self.cks.append(b) if len(self.cks) 2: self.stage TAIL elif self.stage TAIL: if b in (0x0D, 0x0A): self.stage DONE elif self.stage DONE: raw bytes(self.buffer), bytes(self.cks) self.reset() return raw return None这个版本把CRLF中的CR当作结束条件LF会被当作下一帧的第一个无效字节而丢弃若设备只发CR不跟LF同样能结束。帧结束后的状态机会把缓冲区清零进入下一帧等待。该状态机不在这里判断校验和校验留给更高层的解析函数避免状态机里混入业务逻辑。注意BODY阶段里出现的长度前缀在这里不适用必须依靠ETX定位。如果仪器在文本里出现二进制字符状态机就会错乱但对于E1381协议文本只允许可打印ASCII字符这一限制已经被标准规定住了。下一节会在这个状态机之上构造可操作的帧对象。3. 用Python把ASTM E1381帧构造与E1394记录编组串起来3.1 帧类build与parse的对称实现先写一个同时负责编码和解码的帧类把帧序号、文本、校验和三者绑到一起。这样一个对象既能发也能收测试时也方便做往返比对。class AstmFrame: def __init__(self, seq_no: int, text: bytes): self.seq_no seq_no % 8 self.text text def build(self, include_stxFalse) - bytes: body str(self.seq_no).encode(ascii) b\x02 self.text b\x03 cks compute_checksum(b\x02 body if include_stx else body) return b\x02 body cks.encode(ascii) b\r\n classmethod def parse(cls, raw: bytes): if not raw.startswith(b\x02): raise ValueError(缺少帧首STX) body_end raw.find(b\x03) if body_end -1: raise ValueError(缺少ETX) cks raw[body_end1:body_end3] body raw[1:body_end1] # 帧序号到ETX if compute_checksum(body) ! cks.decode(ascii): raise ValueError(f校验和不匹配: {cks}) seq_no raw[1] - 0x30 text raw[3:body_end] # 跳过STX、序号、内层STX return cls(seq_no, text)build里先把帧序号与内层STX拼进body再算校验和。parse则从原始串口帧中取出body先验证校验和再拆帧序号和文本。这里默认文本里不再含有额外STX如果仪器在文本中把字段转义成其他写法需要先用转义处理函数过滤。关于帧序号协议里限定0到7循环所以__init__里做了一次取模。为什么是模8而不是连续计数字符因为E1381规定的帧序号字段是单字符八进制循环。抓包时若发现序号从0跳到2要意识到你可能漏了1号帧好在帧层校验和不会因此误判。3.2 H/P/O/R/C/L记录的分隔规则E1394的记录是一条以|分字段、以^分组件、以分子组件的ASCII文本。每条记录以CR结尾下一记录紧接着出现。常见的记录类型有H头、P患者、O检验申请、R检验结果、C注释、L结尾。为了不被标准条文卡住建议先按“字段位”来理解再对照具体设备手册。下面这段字符串是按常见E1394方言构造的最小结果消息h H|\\^|||DEMO_INSTRUMENT|||||LIS||||||||||20250115103000 p P|1||PA001^^^HOSPITAL^WARD||Zhang^San||19800101|M o O|1|ORD20250115001||^GLU^Glucose||||||||||||O r R|1|^GLU|5.6|mmol/L|3.9-6.1|N||F||||||||||20250115103100 l L|1|N records [h, p, o, r, l] text b.join(rec.encode(ascii) b\r for rec in records) frame AstmFrame(1, text)这里值得注意三点。第一P记录的患者ID区放在第三个字段字段为空时要连续写|||。第二O记录里的检验项目用^GLU^Glucose表示通用代码与名称实际设备在这个项目下还会追加本地代码。第三R记录作为结果记录第4、5、6字段分别是结果值、单位、参考区间解析时够用但未必够全要多看设备实现。3.3 记录编组的两个常见习惯一个工程设计上的问题是让帧层知道H/P/O/R/L吗我的建议是不要。帧层只负责把bytes包进帧记录编组由消息层完成。消息层拿到业务数据后先按字段顺序拼出记录列表再拼成帧正文。各个厂商对字段位的填充习惯不同所以编组函数要接受一个额外的“方言配置”例如哪些字段必须补空位。def build_order_and_result(patient, order, test, value, unit, ref): h fH|\\^|||MEDISYS|||LIS||||||||| p fP|1||{patient[id]}||{patient[name]}||{patient[birth]}|{patient[sex]} o fO|1|{order[id]}||^{test[code]}^{test[name]}||||||||||||O r fR|1|^{test[code]}|{value}|{unit}|{ref}|N||F l L|1|N text \r.join([h, p, o, r, l]).encode(ascii) b\r return text为什么结果消息要以L记录收尾因为接收方要靠L记录判断这次传输的文本是否结束。有的设备还会在EOT之前发多个帧帧序号递增文本里每帧都包含L记录接收方就要按帧序号拼接。不要把L记录当成会话结束会话结束标志是EOT控制字符。这一段既是本地消息层的约定也是对接其他数据库系统时的常见误区。提示H记录里的日期时间在部分仪器上是“最近校准时间”而不是发送时间用于关注仪器的健壮性时要手动核对。4. 通过pyserial把ASTM E1381/1394接到真实串口4.1 串口参数与超时设置RS-232不是即插即用波特率、数据位、校验位和停止位都必须与设备侧一致。大多数支持E1381/1394的仪器使用9600、8、N、1也有老设备用19200或4800。Python这边用pyserial安装方式是在有Python环境后执行python -m pip install pyserial。import serial port serial.Serial( COM3, # Linux为/dev/ttyS1或/dev/ttyUSB0 baudrate9600, bytesizeserial.EIGHTBITS, parityserial.PARITY_NONE, stopbitsserial.STOPBITS_ONE, timeout2.0, # 超过2秒没有新字节就返回空 write_timeout2.0 )这个配置里超时非常关键。E1381的ENQ/ACK交互中若仪器一直不响应read()会阻塞到超时时间。timeout设太短会频繁产生空读取设太长则可能把整个传输卡住。常见经验是把读超时设为“最长两帧间隔的两倍”即仪器连续帧之间的时间间隔乘以2。若无法预先知道先设2秒联调时再用日志优化。4.2 仪器主动发起时的ENQ/ACK与帧应答流程多数仪器不会一开机就发数据而是先发ENQ询问LIS是否就绪。LIS收到ENQ后回ACK仪器随后发一帧或多帧每帧到来都要回ACK/NAK。所有帧收完仪器发EOT结束本次会话。下面这段代码把这套交互放在一个循环里def run_instrument_session(port: serial.Serial, handler): rx FrameReceiver() port.reset_input_buffer() while True: start port.read(1) if not start: continue if start ! b\x05: if start b\x04: return continue port.write(b\x06) while True: byte port.read(1) if not byte: return if byte[0] 0x04: return got rx.feed(byte[0]) if got is None: continue body, checksum got ok compute_checksum(body b\x03) checksum.decode(ascii, errorsignore) port.write(b\x06 if ok else b\x15) if ok: handler(body)这里rx.feed返回的body不包含ETX所以在校验时要手动补回b\x03。checksum是设备发来的两个十六进制字符直接用字符串比较。校验通过就回ACK失败就回NAK。注意EOT在IDLE状态下会被直接丢掉所以当会话结束时循环会退出如果EOT在帧中间出现字节会被当作文本内容最后校验和出错这是上一种做法更耗费时间但不会误吞数据。4.3 LIS主动查询时的发起方写法如果仪器是被动模式LIS作为主站需要主动发起。流程变成LIS发ENQ仪器回ACKLIS发送一条E1394查询帧仪器回ACK随后仪器发送结果数据。发起方代码和上面的差异很小区别只在发送帧后等待对方ACK而不是收到帧后回ACK。def send_query(port: serial.Serial, frame: bytes): port.reset_input_buffer() port.write(b\x05) if port.read(1) ! b\x06: raise TimeoutError(设备未响应ENQ) port.write(frame) ack port.read(1) if ack ! b\x06: raise RuntimeError(f设备返回NAK: {ack!r})这里出现的坑是ACK可能被串口缓存拆分理论上read(1)就能读到一个字节但某些USB转串口线会把多个字节合并返回。稳妥做法是把read(1)换成一个带循环的read_expect_len(port, 1, timeout2)内部循环读取直到凑满长度。这个小函数在对接USB-RS232模块时几乎必写。5. 回归测试用pytest把ASTM E1381/1394行为固定住5.1 帧构建与解析的往返测试联调时如果每次都接真机效率太低。先写一组pytest用例把构造和解析的往返关系固定下来后面改代码时能立刻发现破坏。先为上一节的build和parse设计用例。def test_frame_roundtrip(): text bH|\\^|||DEMO\rP|1||PA001^^^^||Doe||19800101|M\rL|1|N frame AstmFrame(5, text) raw frame.build() parsed AstmFrame.parse(raw) assert parsed.seq_no 5 assert parsed.text text这个用例的价值在于只要build和parse不对称测试马上失败。很多现场bug都来自校验和范围不一致所以再补一个专门校验校验和字段的用例。def test_checksum_known_pattern(): body b1\x02ABC\x03 assert compute_checksum(body) 32 # 0x310x020x410x420x430x035.2 用损坏帧验证NAK逻辑人工构造一个校验和错误的帧确认解析器会抛异常这样串口代码里的NAK分支才会被覆盖。测试时故意把帧文本中的某个字母改掉但保留原校验和。def test_bad_checksum_rejected(): raw AstmFrame(0, bH|\\^|).build() corrupted raw[:3] bX raw[4:] import pytest with pytest.raises(ValueError, match校验和): AstmFrame.parse(corrupted)这组测试看起来简单真正运行后会发现厂商对“检查的字节范围是否包含STX”的做法导致的最多。因此在你自己的项目里可以给AstmFrame增加一个模块级变量CHECKSUM_INCLUDE_STX False测试时就分别用两种模式跑一遍确保切换配置时不会影响解析。5.3 串口层打桩不接设备也能联调收发串口层不建议直接用pytest测真实COM口而要用loopback串口对。Windows可以安装虚拟串口驱动生成COM5/COM6Linux用socat创建成对的pty。创建好后让一个进程往COM5写另一个进程从COM6读。测试代码里把run_instrument_session的handler替换成一个只做断言的函数。def test_ack_nak_path_with_loopback(): # 假定loopback串口已由外部工具创建 s1 serial.Serial(COM5, timeout0.5) s2 serial.Serial(COM6, timeout0.5) s1.write(b\x05) assert s2.read(1) b\x05 s2.write(b\x06) assert s1.read(1) b\x06 s1.close() s2.close()这样即使没有物理设备也能对握手代码做基础验证。真正接仪器前先用真机发一路全流程日志。6. 收尾现场调ASTM E1381/1394的三个实用技巧6.1 双通道hexdump日志一条命令看出帧边界调ASTM最怕的是傻等。我一般会在串口层的每个read后同时打印hex和ascii两行一行原始字节一行可读字符。这样ENQ、ACK、EOT这些控制字符一眼就能分辨不会和文本里的换行混淆。import binascii def log_byte(b: int): logging.info(RX hex%s ascii%r, f{b:02X}, chr(b) if 32 b 126 else )这个技巧比单纯打印pyserial返回的bytes强很多因为REPL直接显示会隐藏控制字符。联调第一件事就是把所有你看到的帧日志保存成文件留作与设备手册逐字节比对。6.2 设备手册与帧日志冲突时先怀疑校验和范围现场最常见的现象是设备发送能进但解析后报校验和错误或者设备收到回复后一直重发同一帧。此时不要急着改校验和算法先把原始帧的校验和计算一遍看它等于“包含STX”还是“不包含STX”的结果。只要加上一个include_stx开关多数设备能立刻匹配。另一类冲突是ETX后只有两个十六进制字符后面没有CR只有LF状态机里要兼容“ETX后跟两个字符后直接遇到LF也能结束”的写法。6.3 用虚拟串口对解决“没有两台电脑”的联调尴尬没有真机时用socat -d -d PTY,link/tmp/ttyA PTY,link/tmp/ttyB生成一对虚拟串口一端跑模拟器一端跑LIS代码几乎能模拟真实收发的所有异常路径。需要特别注意虚拟串口与真实串口在时序上会有细微差别真实设备可能在ACK后等很长时间才发下一帧所以不要把超时参数在虚拟环境下压得太紧。本文还有配套的精品资源点击获取