简介: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,也就是<STX>2<STX>H|\^&|...<ETX>41<CR><LF>。第一个STX表示一帧开始;随后是帧序号,单字符0到7,每发送一帧循环加1;然后又是STX;随后是文本,即E1394的记录内容;到ETX结束;再跟两个十六进制字符作为校验和;最后是CR LF。
ASTM E1381只关心这个外壳,E1394则规定文本里的记录。这样分层的好处是,帧层可以复用,记录层的字段变更不会动摇底层收发。在真正写代码前,先把控制字符的作用整理成一张表。
| 控制字符 | ASCII码 | 本协议中的角色 |
|---|---|---|
| STX | 0x02 | 帧开始,也作为帧序号与记录文本之间的分隔 |
| ETX | 0x03 | 记录文本结束,校验和从这里之后开始 |
| CR | 0x0D | 帧结束,同时也常作为记录分隔符 |
| LF | 0x0A | 帧结束的跟随符,有的设备省略 |
| ENQ | 0x05 | 请求建立传输,相当于“我要开始发了” |
| ACK | 0x06 | 对ENQ或一帧数据的确认 |
| NAK | 0x15 | 否定应答,常见原因是校验和错误 |
| EOT | 0x04 | 本次传输会话结束 |
帧序号不能从帧文本中读出来,而要在状态机里识别。文本部分并不包含帧序号,也不包含第二个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_stx=False) -> 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_end+1:body_end+3] body = raw[1:body_end+1] # 帧序号到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 = f"H|\\^&|||MEDISYS|||LIS|||||||||" p = f"P|1||{patient['id']}||{patient['name']}||{patient['birth']}|{patient['sex']}" o = f"O|1|{order['id']}||^{test['code']}^{test['name']}||||||||||||O" r = f"R|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 baudrate=9600, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=2.0, # 超过2秒没有新字节就返回空 write_timeout=2.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", errors="ignore") 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,仪器回ACK;LIS发送一条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, timeout=2),内部循环读取直到凑满长度。这个小函数在对接USB-RS232模块时几乎必写。
5. 回归测试:用pytest把ASTM E1381/1394行为固定住
5.1 帧构建与解析的往返测试
联调时如果每次都接真机,效率太低。先写一组pytest用例,把构造和解析的往返关系固定下来,后面改代码时能立刻发现破坏。先为上一节的build和parse设计用例。
def test_frame_roundtrip(): text = b"H|\\^&|||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 = b"1\x02ABC\x03" assert compute_checksum(body) == "32" # 0x31+0x02+0x41+0x42+0x43+0x035.2 用损坏帧验证NAK逻辑
人工构造一个校验和错误的帧,确认解析器会抛异常,这样串口代码里的NAK分支才会被覆盖。测试时故意把帧文本中的某个字母改掉,但保留原校验和。
def test_bad_checksum_rejected(): raw = AstmFrame(0, b"H|\\^&|").build() corrupted = raw[:3] + b"X" + 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/COM6,Linux用socat创建成对的pty。创建好后让一个进程往COM5写,另一个进程从COM6读。测试代码里把run_instrument_session的handler替换成一个只做断言的函数。
def test_ack_nak_path_with_loopback(): # 假定loopback串口已由外部工具创建 s1 = serial.Serial("COM5", timeout=0.5) s2 = serial.Serial("COM6", timeout=0.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后等很长时间才发下一帧,所以不要把超时参数在虚拟环境下压得太紧。
本文还有配套的精品资源,点击获取