실습 3 — 화면을 API 와 잇기

import json, os, subprocess, sys, time, urllib.request
from pathlib import Path

# 13차시 실습 4처럼 uvicorn 으로 띄운다. 주소는 127.0.0.1 로 적는다(13차시)
api = subprocess.Popen([sys.executable, "-m", "uvicorn", "out_delay_api:app", "--port", "8013", "--log-level", "warning"])
for _ in range(60):
    try:
        print(urllib.request.urlopen("http://127.0.0.1:8013/health").read().decode())
        break
    except OSError:
        time.sleep(0.5)

Path("out_app_api.py").write_text('''"""배송 지연 예측 화면(API 연결). 실행: streamlit run out_app_api.py"""
import json
import os
import urllib.error
import urllib.request

import streamlit as st

API = os.environ.get("API_URL", "http://127.0.0.1:8013")       # 어디서 띄우느냐에 따라 바뀌는 주소


def call(path, body=None, timeout=5):
    data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
    req = urllib.request.Request(API + path, data=data, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:    # 시간 제한을 꼭 준다
        return json.loads(r.read())


@st.cache_data(ttl=60)                         # 보기 목록은 1분 동안 저장해 두고 다시 묻지 않는다
def choices():
    props = call("/openapi.json")["components"]["schemas"]["Order"]["properties"]
    return {k: v["enum"] for k, v in props.items() if "enum" in v}


st.title("배송 지연 예측")
try:
    health = call("/health")
    opts = choices()
except OSError:                                # 연결 거부, 시간 초과
    st.error(f"API 서버({API})에 연결할 수 없습니다. 서버가 켜져 있는지 확인하세요.")
    st.stop()                                  # 여기서 스크립트를 멈춘다. 아래 칸은 그리지 않는다
st.caption(f"API 연결됨 · 모델 판 {health['판']}")

with st.form("order"):
    order = {k: st.selectbox(k, v, key=k) for k, v in opts.items()}
    order["주문시각"] = st.slider("주문시각", 0, 23, 14, key="주문시각")
    order["거리km"] = st.number_input("거리km", 0.5, 1000.0, 20.0, key="거리km")
    order["무게kg"] = st.number_input("무게kg", 0.0, 100.0, 2.0, key="무게kg")
    sent = st.form_submit_button("예측하기")

if sent:
    try:
        result = call("/predict", order)
    except urllib.error.HTTPError as e:        # API 가 4xx, 5xx 로 거절했을 때
        st.error(f"요청이 거절됐습니다({e.code}).")
        st.stop()
    st.metric("지연 확률", f"{result['지연확률']:.1%}")
    (st.warning if result["지연예상"] else st.success)("지연 예상" if result["지연예상"] else "제때 도착 예상")
''', encoding="utf-8")
print("화면 파일:", Path("out_app_api.py").stat().st_size, "바이트")

from streamlit.testing.v1 import AppTest

os.environ["API_URL"] = "http://127.0.0.1:8013"
at = AppTest.from_file(str(Path("out_app_api.py").resolve()), default_timeout=60).run()
print(at.caption[0].value)
print("API 에서 읽어 온 보기:", {s.label: s.options for s in at.selectbox})
for key, value in {"지역": "지방", "상품분류": "도서", "날씨": "눈", "택배사": "빠른별택배"}.items():
    at.selectbox(key=key).set_value(value)
at.slider(key="주문시각").set_value(22)
at.number_input(key="거리km").set_value(197.6)
at.number_input(key="무게kg").set_value(4.6)
at.button[0].click().run()
print(at.metric[0].value, [w.value for w in at.warning])

api.terminate()
api.wait(timeout=10)
at2 = AppTest.from_file(str(Path("out_app_api.py").resolve()), default_timeout=60).run()
print("오류 알림:", [e.value for e in at2.error])
print("그려진 선택 상자 수:", len(at2.selectbox))