6. 연동펌프 일곱 — 한 줄에 주소 일곱 · 유박사의 mRNA 백신 합성기 펌웨어 첫걸음

연동펌프가 일곱 대입니다. 일곱 대에 선을 각각 끌면 배선이 감당이 안 됩니다. 그래서 RS485 를 씁니다 — 선 두 가닥에 일곱 대를 전부 매답니다.

대신 “누구에게 하는 말인지” 를 밝혀야 합니다. 그것이 슬레이브 번호(1~247)이고, 그 위에서 도는 약속이 Modbus RTU 입니다.

import minimalmodbus, serial

pump = minimalmodbus.Instrument('/dev/pump-bus', 3)   # 3번 펌프
pump.serial.baudrate = 9600
pump.serial.timeout  = 0.3
pump.write_register(REG_RPM, 1500, 0)

펌프 수와 통신 설정을 바꿔 보세요. 한 바퀴 도는 데 걸리는 시간이 예산을 넘는지 보입니다.

불러오는 중…

한 줄을 여럿이 나눠 쓰면 시간이 예산이 됩니다. 이 계산을 스케줄러 설계 전에 해 둡니다.

Python 브라우저에서 실행

# RS485 는 한 줄을 '차례로' 쓴다. 일곱 대를 돌아보는 데 얼마나 걸리나?
baud = 9600
bits_per_char = 11              # 시작 1 + 데이터 8 + 정지 1 + 여유
char_time = bits_per_char / baud

# Modbus RTU 한 번 주고받기: 요청 8바이트 + 응답 7바이트 + 침묵 3.5문자 × 2
def turnaround(n_req=8, n_rsp=7, silent=3.5):
    return (n_req + n_rsp + silent * 2) * char_time

t1 = turnaround()
print("한 대와 한 번 주고받기: %.1f ms" % (t1 * 1000))
for n in (7, 8):
    print("  %d 대를 한 바퀴: %.0f ms" % (n, t1 * n * 1000))
print()

print("%-10s %-14s %-14s" % ("보율", "한 대", "여덟 대 한 바퀴"))
for b in (9600, 19200, 38400, 115200):
    ct = bits_per_char / b
    t = (8 + 7 + 7) * ct
    print("%8d %10.1f ms %12.0f ms" % (b, t * 1000, t * 8 * 1000))
print()
print("스케줄러가 1초마다 모든 펌프 상태를 확인하려 한다면,")
print("9600 보율에서 여덟 대는 이미 예산의 상당 부분을 먹는다.")
print("→ '필요할 때만 말하고, 상태는 드물게 묻는다' 가 답이다. (18강)")

리눅스 쪽에서 해 둘 일 넷

할 일	까닭
`sudo usermod -aG dialout $USER`	포트를 열 권한. `sudo python` 으로 때우지 말 것
udev 규칙으로 `/dev/pump-bus` 고정	`ttyUSB0` 번호는 다시 켤 때마다 바뀐다
`systemctl disable --now ModemManager`	이것이 포트를 먼저 잡아 첫 통신을 망친다
양 끝에만 120 Ω 종단	중간에 달거나 별 모양으로 배선하면 간헐적 CRC 오류

그리고 코드 쪽에서 하나 — 버스는 하나이므로 접근을 하나로 묶어야 합니다. 일곱 개 객체가 제각기 포트를 열면 서로를 밟습니다. 자물쇠(lock) 하나 뒤에 모든 통신을 세워 두십시오.

레지스터 표를 코드가 아니라 파일에 둡니다. 펌프를 바꿔도 코드는 그대로입니다.

Python 브라우저에서 실행

# 장치 프로필 — 펌프마다 다른 레지스터 표를 '파일' 로 뺀다.
# 코드는 그대로 두고 파일만 갈아 끼우면 다른 펌프를 쓴다.
profile = {
    "model": "우리 실험실 펌프 A",
    "baudrate": 9600,
    "registers": {
        "rpm":       {"addr": 0x0001, "scale": 10,  "rw": "rw"},
        "direction": {"addr": 0x0002, "scale": 1,   "rw": "rw"},
        "run_stop":  {"addr": 0x0003, "scale": 1,   "rw": "rw"},
        "status":    {"addr": 0x0010, "scale": 1,   "rw": "r"},
    },
}

class PumpProfile:
    def __init__(self, prof): self.p = prof
    def encode(self, name, value):
        r = self.p["registers"][name]
        if "w" not in r["rw"]:
            raise PermissionError("%s 는 읽기 전용이다" % name)
        return r["addr"], int(round(value * r["scale"]))

pp = PumpProfile(profile)
for name, val in [("rpm", 150.0), ("run_stop", 1)]:
    addr, raw = pp.encode(name, val)
    print("%-10s → 레지스터 0x%04X 에 %d 쓰기" % (name, addr, raw))

try:
    pp.encode("status", 1)
except PermissionError as e:
    print("\n막혔다:", e)

🤔 확인 문제

펌프 일곱 대를 RS485 한 줄에 매달 때 코드에서 반드시 해야 하는 것은?

🤔 확인 문제

이 강의가 펌프 레지스터 주소를 적어 주지 않는 까닭은?