ant-vision-inspector/find_cells_trigger.py

"""
In-Sight 2001C-353 트리거 후 전체 셀 스캔 스크립트
SE8 트리거 실행 → 결과 갱신 완료 후 A0~Z50 범위 GV 스캔.
결과는 콘솔 출력과 find_cells_trigger_result.txt 에 동시 저장.
"""

import socket
import string
import time

IP   = "169.254.0.1"
PORT = 23
COLS = list(string.ascii_uppercase)   # A ~ Z (26개)
ROWS = range(0, 251)                  # 0 ~ 250 (251개) → 총 6526셀
RESULT_FILE = "find_cells_trigger_result.txt"


# ------------------------------------------------------------------ #
# 저수준 소켓 헬퍼 (insight.py 방식 동일)
# ------------------------------------------------------------------ #

def _read_line(sock: socket.socket, buf: bytearray) -> str:
    """버퍼에서 \\r\\n까지 읽어 문자열로 반환. 남은 데이터는 buf에 보존."""
    while b"\r\n" not in buf:
        try:
            chunk = sock.recv(4096)
        except socket.timeout:
            break
        if not chunk:
            break
        buf += chunk

    if b"\r\n" in buf:
        idx  = buf.index(b"\r\n")
        line = bytes(buf[:idx])
        del buf[:idx + 2]
    else:
        line = bytes(buf)
        buf.clear()

    for encoding in ["euc-kr", "cp949", "utf-8"]:
        try:
            return line.decode(encoding).strip()
        except Exception:
            continue
    return line.decode(errors="ignore").strip()


def _send(sock: socket.socket, cmd: str):
    sock.sendall((cmd + "\r\n").encode())


# ------------------------------------------------------------------ #
# 로그인 (insight.py connect() 시퀀스와 동일)
# ------------------------------------------------------------------ #

def _login(sock: socket.socket, buf: bytearray):
    banner = _read_line(sock, buf)
    print(f"[연결] 배너: {banner!r}")

    _send(sock, "admin")
    while b"Password:" not in buf:
        try:
            chunk = sock.recv(4096)
        except socket.timeout:
            break
        if not chunk:
            break
        buf += chunk
    prompt = buf.decode(errors="ignore").strip()
    buf.clear()
    print(f"[연결] 프롬프트: {prompt!r}")

    sock.sendall(b"\r\n")              # 빈 비밀번호
    result = _read_line(sock, buf)
    print(f"[연결] 로그인: {result!r}")

    if "Logged In" not in result:
        raise RuntimeError(f"로그인 실패: {result!r}")


# ------------------------------------------------------------------ #
# SE8 트리거
# ------------------------------------------------------------------ #

def _trigger(sock: socket.socket, buf: bytearray):
    """SE8 소프트웨어 트리거 전송 후 응답 수신."""
    _send(sock, "SE8")
    response = _read_line(sock, buf)
    print(f"[트리거] SE8 응답: {response!r}")


# ------------------------------------------------------------------ #
# 셀 스캔
# ------------------------------------------------------------------ #

def _scan(sock: socket.socket, buf: bytearray) -> dict:
    """GV[열][행] 전체 스캔 → {셀주소: 값문자열}"""
    total = len(COLS) * len(ROWS)
    found = {}

    for col in COLS:
        for row in ROWS:
            cell = f"{col}{row}"
            _send(sock, f"GV{cell}")

            status = _read_line(sock, buf)   # "1" 또는 "0"

            if status == "1":
                value = _read_line(sock, buf)
                if not value.strip():
                    pass
                else:
                    found[cell] = value
                    try:
                        print(f"\r[{cell:<4}] = {value:<40}")
                    except UnicodeEncodeError:
                        print(f"\r[{cell:<4}] = {value.encode('utf-8', errors='replace')!r}")

            done = (COLS.index(col)) * len(ROWS) + row + 1
            print(
                f"스캔 중... [{cell} / Z{ROWS[-1]}]  ({done}/{total})",
                end="\r", flush=True,
            )

    print()   # 진행바 줄 정리
    return found


# ------------------------------------------------------------------ #
# 결과 저장
# ------------------------------------------------------------------ #

def _save(found: dict, path: str = RESULT_FILE):
    with open(path, "w", encoding="utf-8") as f:
        f.write(f"스캔 범위: A0 ~ Z{ROWS[-1]}  /  유효 셀: {len(found)}개\n")
        f.write("=" * 40 + "\n")
        for cell, val in found.items():
            f.write(f"[{cell:<4}] = {val}\n")
    print(f"결과 저장 완료 → {path}")


# ------------------------------------------------------------------ #
# 값 검색
# ------------------------------------------------------------------ #

def _search(found: dict, query: str):
    matches = {c: v for c, v in found.items() if query.lower() in v.lower()}
    if matches:
        print(f"\n[검색] '{query}' 포함 셀 {len(matches)}개:")
        for cell, val in matches.items():
            print(f"  [{cell:<4}] = {val}")
    else:
        print(f"[검색] '{query}'를 포함한 셀이 없습니다.")


# ------------------------------------------------------------------ #
# 메인
# ------------------------------------------------------------------ #

def main():
    print(f"[연결] {IP}:{PORT} 접속 중...")
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3.0)
    sock.connect((IP, PORT))
    buf = bytearray()

    try:
        _login(sock, buf)

        input("\n준비됐으면 엔터를 누르세요 (리플렉터를 카메라 앞에 올려두세요): ")

        print("[트리거] SE8 전송 중...")
        _trigger(sock, buf)

        print("[대기] 카메라 처리 완료 대기 (1초)...")
        time.sleep(1)

        print(f"\n스캔 시작: A0 ~ Z{ROWS[-1]}  (총 {len(COLS) * len(ROWS)}셀)\n")
        found = _scan(sock, buf)
    finally:
        sock.close()

    print(f"\n스캔 완료. 총 {len(found)}개 셀 발견")
    print("=" * 40)
    for cell, val in found.items():
        try:
            print(f"  [{cell:<4}] = {val}")
        except UnicodeEncodeError:
            print(f"  [{cell:<4}] = {val.encode('utf-8', errors='replace')!r}")

    _save(found)

    print()
    while True:
        query = input("찾는 값 입력 (엔터 시 종료): ").strip()
        if not query:
            break
        _search(found, query)


if __name__ == "__main__":
    main()