shkim
bc7e52e620
- check_input(): 비동기 키보드 입력 (select 사용) - process_command(): 명령어 처리 - 명령어: c(영점), p(일시정지), r(재시작), h(도움말), q(종료)
212 lines
5.0 KiB
Python
212 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
LoadCell - Raspberry Pi Python library for HX711 load cell amplifier
|
|
|
|
기능 :
|
|
1. 무게 측정 (양수 값으로 출력)
|
|
2. 키보드 명령어: 'c' 영점조절, 'p' 일시정지, 'r' 재시작
|
|
3. 시작 시 자동 캘리브레이션
|
|
|
|
하드웨어 연결:
|
|
- HX711 [VCC] -> Pi 5 [Pin 1] (3.3V)
|
|
- HX711 [GND] -> Pi 5 [Pin 6] (GND)
|
|
- HX711 [DT] -> Pi 5 [Pin 29] (GPIO5)
|
|
- HX711 [SCK] -> Pi 5 [Pin 31] (GPIO6)
|
|
"""
|
|
|
|
from HX711 import SimpleHX711, Rate
|
|
import time
|
|
import sys
|
|
import select
|
|
import tty
|
|
import termios
|
|
|
|
# HX711 핀 설정
|
|
DT_PIN = 5 # GPIO5 -> 29번 핀
|
|
SCK_PIN = 6 # GPIO6 -> 31번 핀
|
|
|
|
# 설정 변수
|
|
CALIBRATION_FACTOR = -411.17 # 캘리브레이션 팩터
|
|
measure_delay = 0.1 # 측정 간격 (초)
|
|
density = 1.0 # 기본값: 물 (1.0 g/ml)
|
|
|
|
# 상태 변수
|
|
is_paused = False
|
|
is_calibrated = False
|
|
offset_value = 0
|
|
|
|
# HX711 인스턴스
|
|
hx = None
|
|
|
|
|
|
def init_hx711():
|
|
"""HX711 초기화"""
|
|
global hx
|
|
print("HX711 초기화 중...")
|
|
print(f" DT 핀: GPIO{DT_PIN} (Pin 29)")
|
|
print(f" SCK 핀: GPIO{SCK_PIN} (Pin 31)")
|
|
|
|
try:
|
|
hx = SimpleHX711(DT_PIN, SCK_PIN, 1, 0, Rate.HZ_10)
|
|
print("HX711 연결 성공!")
|
|
return True
|
|
except Exception as e:
|
|
print(f"HX711 연결 실패: {e}")
|
|
print("배선 확인 필요")
|
|
return False
|
|
|
|
|
|
def perform_calibration():
|
|
"""캘리브레이션 수행"""
|
|
global is_calibrated, offset_value
|
|
|
|
if is_calibrated:
|
|
return
|
|
|
|
print()
|
|
print("-" * 50)
|
|
print(">> 최초 캘리브레이션 진행")
|
|
print("-" * 50)
|
|
print("로드셀 위에 아무것도 올리지 마세요!")
|
|
print("2초 후 영점 조정을 시작합니다...")
|
|
time.sleep(2)
|
|
|
|
print("영점 조정 중...")
|
|
|
|
readings = []
|
|
for _ in range(10):
|
|
readings.append(hx.read())
|
|
time.sleep(0.1)
|
|
|
|
offset_value = sum(readings) / len(readings)
|
|
|
|
print(">> 캘리브레이션 완료!")
|
|
print(f"현재 오프셋 값: {offset_value:.2f}")
|
|
print()
|
|
|
|
is_calibrated = True
|
|
|
|
|
|
def tare():
|
|
"""영점 조정 함수"""
|
|
global offset_value
|
|
|
|
print("\n>> 영점 조정 중...")
|
|
print(" 로드셀 위에 아무것도 올리지 마세요")
|
|
time.sleep(1)
|
|
|
|
readings = []
|
|
for _ in range(10):
|
|
readings.append(hx.read())
|
|
time.sleep(0.1)
|
|
|
|
offset_value = sum(readings) / len(readings)
|
|
print(">> 영점 조정 완료!\n")
|
|
|
|
|
|
def get_weight():
|
|
"""무게 측정 함수"""
|
|
raw_value = hx.read()
|
|
weight_g = (raw_value - offset_value) / CALIBRATION_FACTOR
|
|
|
|
volume_ml = weight_g / density
|
|
return weight_g, volume_ml
|
|
|
|
|
|
def print_help():
|
|
"""도움말 출력"""
|
|
print()
|
|
print("-" * 50)
|
|
print("키보드 명령어:")
|
|
print(" 'c' : 영점 조절 (Calibration)")
|
|
print(" 'p' : 일시정지 (Pause)")
|
|
print(" 'r' : 재시작 (Resume)")
|
|
print(" 'h' : 도움말 보기")
|
|
print(" 'q' : 종료 (Quit)")
|
|
print("-" * 50)
|
|
print()
|
|
|
|
|
|
def check_input():
|
|
"""비동기 키보드 입력 확인"""
|
|
if select.select([sys.stdin], [], [], 0)[0]:
|
|
return sys.stdin.read(1)
|
|
return None
|
|
|
|
|
|
def process_command(cmd):
|
|
"""명령어 처리"""
|
|
global is_paused
|
|
|
|
cmd = cmd.lower()
|
|
|
|
if cmd == 'c':
|
|
tare()
|
|
|
|
elif cmd == 'p':
|
|
if not is_paused:
|
|
is_paused = True
|
|
print("\n>> 측정 일시정지됨")
|
|
print(" 'r'을 입력하면 다시 시작합니다.\n")
|
|
else:
|
|
print("\n>> 이미 일시정지 상태입니다.\n")
|
|
|
|
elif cmd == 'r':
|
|
if is_paused:
|
|
is_paused = False
|
|
print("\n>> 측정 재시작!\n")
|
|
else:
|
|
print("\n>> 이미 측정 중입니다.\n")
|
|
|
|
elif cmd == 'h':
|
|
print_help()
|
|
|
|
elif cmd == 'q':
|
|
print("\n>> 프로그램 종료\n")
|
|
return False
|
|
|
|
elif cmd not in ['\n', '\r', ' ']:
|
|
print(f">> 알 수 없는 명령어: {cmd}")
|
|
print(" 'h'를 입력하면 도움말을 볼 수 있습니다.")
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
if not init_hx711():
|
|
return
|
|
|
|
perform_calibration()
|
|
print_help()
|
|
|
|
old_settings = termios.tcgetattr(sys.stdin)
|
|
|
|
try:
|
|
tty.setcbreak(sys.stdin.fileno())
|
|
print("명령어 테스트 (q: 종료)...\n")
|
|
|
|
while True:
|
|
cmd = check_input()
|
|
if cmd:
|
|
if not process_command(cmd):
|
|
break
|
|
|
|
if is_paused:
|
|
time.sleep(0.1)
|
|
continue
|
|
|
|
weight_g, volume_ml = get_weight()
|
|
print(f"측정: {weight_g:.1f}g / {volume_ml:.1f}ml")
|
|
time.sleep(measure_delay)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n>> 프로그램 종료\n")
|
|
|
|
finally:
|
|
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
|
|
print("프로그램 종료됨.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|