chan/ocr_macro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

섹션 조회 추가

Browse Source
This commit is contained in:
조찬영
2025-09-03 10:29:02 +09:00
parent 443b2cc82e
commit eed8723f51
6 changed files with 493 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
docker-compose.yml
View File

@@ -4,11 +4,23 @@ version: '3.8'
services:
ui:
image: ocr-comparison-ui
volumes:
- ./workspace:/workspace
build:
context: .
dockerfile: dockerfile # Dockerfile의 상대 경로를 직접 지정
dockerfile: dockerfile
volumes:
- ./workspace:/workspace
ports:
- "8501:8501"
command: streamlit run app.py --server.port=8501
eval_ui:
image: ocr-comparison-ui
build:
context: .
dockerfile: dockerfile
volumes:
- ./workspace:/workspace
ports:
- "8601:8601"
command: streamlit run ocr_eval_app.py --server.port=8601

17
dockerfile
View File

@@ -1,17 +1,18 @@
# Dockerfile
FROM python:3.10-slim
RUN pip install uv
WORKDIR /workspace
COPY requirements.txt .
RUN uv pip install --system --no-cache -r requirements.txt
COPY workspace/ .
RUN pip install --no-cache-dir -r requirements.txt
COPY workspace/app.py .
COPY . .
# 서비스가 사용할 포트를 명시합니다 (문서화 목적).
EXPOSE 8501
EXPOSE 8502
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]
# 컨테이너의 기본 실행 명령을 정의합니다.
# 이 명령은 docker-compose.yml 파일의 command에 의해 각 서비스별로 재정의됩니다.
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

8
requirements.txt
View File

@@ -2,4 +2,10 @@
streamlit
requests
python-dotenv
pandas
pandas
fuzzywuzzy==0.18.0
jiwer==4.0.0
# levenshtein and python-levenshtein are replaced by rapidfuzz for better compatibility
rapidfuzz==3.13.0
markupsafe==3.0.2
werkzeug==3.1.3

78
workspace/app.py
View File

@@ -44,6 +44,13 @@ def save_completed_file(seed, basename):
json.dump(list(completed_set), f, indent=2)
def get_existing_sessions():
""" "shared_sessions" 디렉토리에서 기존 세션 목록을 가져옵니다. """
if not SESSION_BASE_PATH.exists():
return []
return sorted([d.name for d in SESSION_BASE_PATH.iterdir() if d.is_dir()])
# --- 헬퍼 함수 ---
@@ -196,26 +203,65 @@ def main():
st.session_state.current_index = 0
SESSION_BASE_PATH.mkdir(parents=True, exist_ok=True)
matched_files = None
st.sidebar.info("화면을 넓게 보려면 오른쪽 위 화살표를 누르세요 <<")
st.sidebar.markdown("---")
st.sidebar.header("세션 선택")
existing_sessions = get_existing_sessions()
session_options = ["새 세션 생성"] + existing_sessions
current_seed_from_url = st.query_params.get("seed")
# URL에 시드가 없으면 "새 세션 생성"을 기본값으로
if not current_seed_from_url:
current_selection = "새 세션 생성"
# URL의 시드가 존재하지 않는 세션이면 경고 후 "새 세션 생성"으로
elif current_seed_from_url not in existing_sessions:
st.sidebar.warning(f"URL의 시드 '{current_seed_from_url}'에 해당하는 세션을 찾을 수 없습니다.")
current_selection = "새 세션 생성"
# 잘못된 시드는 URL에서 제거
if "seed" in st.query_params:
del st.query_params["seed"]
else:
current_selection = current_seed_from_url
selected_session = st.sidebar.selectbox(
"작업할 세션을 선택하세요.",
session_options,
index=session_options.index(current_selection),
key="session_selector",
)
# 사용자가 선택을 변경하면 URL을 업데이트하고 앱을 다시 실행
if selected_session != current_selection:
if selected_session == "새 세션 생성":
if "seed" in st.query_params:
del st.query_params["seed"]
else:
st.query_params["seed"] = selected_session
st.rerun()
# --- 이후 로직은 URL의 'seed' 쿼리 파라미터를 기반으로 동작 ---
url_seed = st.query_params.get("seed")
matched_files = None
completed_files = set()
if url_seed:
completed_files = load_completed_files(url_seed)
files = load_files_from_session(url_seed)
if files[0] is not None:
st.success(f"'{url_seed}' 시드에서 파일을 불러왔습니다.")
st.success(f"'{url_seed}' 세션에서 파일을 불러왔습니다.")
matched_files = match_files_3_way(*files)
else:
# 이 경우는 위에서 처리되었지만, 안전장치로 남겨둠
st.error(f"'{url_seed}'에 해당하는 세션을 찾을 수 없습니다.")
else:
completed_files = set()
if "seed" in st.query_params:
del st.query_params["seed"]
st.rerun()
# --- 사이드바 ---
st.sidebar.info("화면을 넓게 보려면 오른쪽 위 화살표를 누르세요 <<")
st.sidebar.markdown("---")
st.sidebar.header("파일 업로드")
if not matched_files:
# 파일 업로드 UI는 새 세션 생성 시에만 표시
if not url_seed:
st.sidebar.header("새 세션 생성 (파일 업로드)")
docs = st.sidebar.file_uploader(
"1. 원본 문서", accept_multiple_files=True, type=["png", "jpg", "pdf"]
)
@@ -238,12 +284,14 @@ def main():
st.sidebar.info("URL을 복사하여 다른 사람과 세션을 공유하세요.")
st.sidebar.text_input("공유 시드", url_seed, disabled=True)
if not url_seed:
st.info("새로운 세션을 생성하려면 사이드바에서 모든 종류의 파일을 업로드하세요.")
return
if not matched_files:
st.info("모든 종류의 파일을 업로드하고 세션을 생성하세요.")
if matched_files is not None and not matched_files:
st.warning(
"파일 이름(확장자 제외)이 동일한 '문서-paddle_ocr-upstage' 세트를 찾을 수 없습니다."
)
st.warning(
"파일 이름(확장자 제외)이 동일한 '문서-paddle_ocr-upstage' 세트를 찾을 수 없습니다."
)
return
st.sidebar.header("파일 탐색")

307
workspace/ocr_eval_app.py Normal file
View File

@@ -0,0 +1,307 @@
# ocr_eval_app.py
import base64
import json
import difflib
from pathlib import Path
import pandas as pd
import io
import streamlit as st
from ocr_eval_engine import OCREvaluator
# --- 상수 ---
SESSION_BASE_PATH = Path(__file__).parent / "shared_sessions"
EDIT_KEY = "parsed"
# --- 헬퍼 함수 ---
def get_evaluable_sessions():
""" "shared_sessions"에서 'groundtruth' 폴더를 포함하는 세션 목록을 가져옵니다. """
if not SESSION_BASE_PATH.exists():
return []
evaluable = []
for d in SESSION_BASE_PATH.iterdir():
if d.is_dir() and (d / "groundtruth").is_dir():
evaluable.append(d.name)
return sorted(evaluable)
def get_session_path(seed):
return SESSION_BASE_PATH / seed
def display_pdf(file_path):
"""PDF 파일을 iframe으로 표시합니다."""
bytes_data = file_path.read_bytes()
base64_pdf = base64.b64encode(bytes_data).decode("utf-8")
st.markdown(
f'<iframe src="data:application/pdf;base64,{base64_pdf}" width="100%" height="800" type="application/pdf"></iframe>',
unsafe_allow_html=True,
)
def generate_hyp_html(ref: str, hyp: str) -> str:
"""
difflib.SequenceMatcher를 사용하여 가설(hyp) 텍스트의 오류를 시각화하는 HTML을 생성합니다.
"""
matcher = difflib.SequenceMatcher(None, ref, hyp)
html_out = ""
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
hyp_chunk = hyp[j1:j2]
hyp_chunk_display = hyp_chunk.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
hyp_chunk_display = hyp_chunk_display.replace("\n", "<br>").replace(" ", "&nbsp;")
if tag == 'equal':
html_out += f'<span style="color: black;">{hyp_chunk_display}</span>'
elif tag == 'replace':
html_out += f'<span style="color: red; background-color: #ffdddd; font-weight: bold;">{hyp_chunk_display}</span>'
elif tag == 'insert':
html_out += f'<span style="color: green; background-color: #ddffdd; font-weight: bold;">{hyp_chunk_display}</span>'
return f'<div style="font-family: monospace; border: 1px solid #ddd; padding: 10px; border-radius: 5px; white-space: normal; word-break: break-all; line-height: 1.6;">{html_out}</div>'
def match_evaluation_files(seed):
"""
세션 폴더 내에서 평가에 필요한 파일들의 목록을 찾아서 매칭합니다.
"""
session_path = get_session_path(seed)
doc_path = session_path / "docs"
gt_path = session_path / "groundtruth"
paddle_path = session_path / "jsons_paddle_ocr"
upstage_path = session_path / "jsons_upstage"
if not all([p.is_dir() for p in [doc_path, gt_path, paddle_path, upstage_path]]):
return None
gt_files = {f.stem for f in gt_path.glob("*.json")}
doc_map = {f.stem: f for f in doc_path.iterdir()}
paddle_map = {f.stem: f for f in paddle_path.glob("*.json")}
upstage_map = {f.stem: f for f in upstage_path.glob("*.json")}
matched = {}
for stem in sorted(list(gt_files)):
if stem in doc_map and stem in paddle_map and stem in upstage_map:
matched[stem] = {
"doc_file": doc_map[stem],
"gt_file": gt_path / f"{stem}.json",
"paddle_file": paddle_map[stem],
"upstage_file": upstage_map[stem],
}
return matched
def display_evaluation_for_file(files):
"""선택된 파일에 대한 평가 결과를 표시합니다."""
st.header("📊 성능 평가 결과")
try:
with open(files["gt_file"], "r", encoding="utf-8") as f:
gt_data = json.load(f)
with open(files["paddle_file"], "r", encoding="utf-8") as f:
paddle_data = json.load(f)
with open(files["upstage_file"], "r", encoding="utf-8") as f:
upstage_data = json.load(f)
gt_text = (gt_data[0] if isinstance(gt_data, list) else gt_data).get(EDIT_KEY, "")
paddle_text = (paddle_data[0] if isinstance(paddle_data, list) else paddle_data).get(EDIT_KEY, "")
upstage_text = (upstage_data[0] if isinstance(upstage_data, list) else upstage_data).get(EDIT_KEY, "")
if not gt_text:
st.warning("정답 텍스트가 비어있어 평가할 수 없습니다.")
return
evaluator = OCREvaluator(gt_text)
paddle_results = evaluator.evaluate(paddle_text)
upstage_results = evaluator.evaluate(upstage_text)
col1, col2 = st.columns(2)
with col1:
st.markdown("#### Model 1: Paddle OCR")
m_col1, m_col2 = st.columns(2)
m_col1.metric("엄격한 WER", f"{paddle_results['strict_wer']:.2%}")
m_col2.metric("엄격한 CER", f"{paddle_results['strict_cer']:.2%}")
m_col1.metric("유연한 WER", f"{paddle_results['flexible_wer']:.2%}")
m_col2.metric("유연한 CER", f"{paddle_results['flexible_cer']:.2%}")
with col2:
st.markdown("#### Model 2: Upstage OCR")
m_col1, m_col2 = st.columns(2)
m_col1.metric("엄격한 WER", f"{upstage_results['strict_wer']:.2%}")
m_col2.metric("엄격한 CER", f"{upstage_results['strict_cer']:.2%}")
m_col1.metric("유연한 WER", f"{upstage_results['flexible_wer']:.2%}")
m_col2.metric("유연한 CER", f"{upstage_results['flexible_cer']:.2%}")
with st.expander("상세 텍스트 비교", expanded=True):
st.markdown("""
<style>.legend{display:flex;align-items:center;margin-bottom:10px;}.legend-box{width:20px;height:20px;margin-right:10px;border:1px solid #ccc;}</style>
<b>범례 (Legend)</b>
<div class="legend"><div class="legend-box" style="background-color:white;"></div><span>일치하는 텍스트</span></div>
<div class="legend"><div class="legend-box" style="background-color:#ddffdd;"></div><span><b>삽입된 텍스트</b> (정답에 없음)</span></div>
<div class="legend"><div class="legend-box" style="background-color:#ffdddd;"></div><span><b>치환된 텍스트</b> (정답과 다름)</span></div>
""", unsafe_allow_html=True)
st.markdown("---")
text_col1, text_col2, text_col3 = st.columns(3)
with text_col1:
st.text("정답 (Ground Truth)")
st.code(gt_text, language=None)
with text_col2:
st.text("Paddle OCR")
html = generate_hyp_html(gt_text, paddle_text)
st.markdown(html, unsafe_allow_html=True)
with text_col3:
st.text("Upstage OCR")
html = generate_hyp_html(gt_text, upstage_text)
st.markdown(html, unsafe_allow_html=True)
except Exception as e:
st.error(f"성능 평가 중 오류 발생: {e}")
@st.cache_data
def generate_all_results_df(_matched_files):
"""세션의 모든 파일에 대한 평가 결과를 집계하여 DataFrame으로 반환합니다."""
all_results = []
for basename, files in _matched_files.items():
try:
with open(files["gt_file"], "r", encoding="utf-8") as f:
gt_data = json.load(f)
with open(files["paddle_file"], "r", encoding="utf-8") as f:
paddle_data = json.load(f)
with open(files["upstage_file"], "r", encoding="utf-8") as f:
upstage_data = json.load(f)
gt_text = (gt_data[0] if isinstance(gt_data, list) else gt_data).get(EDIT_KEY, "")
if not gt_text:
continue
evaluator = OCREvaluator(gt_text)
# Paddle 모델 평가
paddle_text = (paddle_data[0] if isinstance(paddle_data, list) else paddle_data).get(EDIT_KEY, "")
paddle_results = evaluator.evaluate(paddle_text)
paddle_results['model'] = 'paddle_ocr'
paddle_results['file'] = basename
all_results.append(paddle_results)
# Upstage 모델 평가
upstage_text = (upstage_data[0] if isinstance(upstage_data, list) else upstage_data).get(EDIT_KEY, "")
upstage_results = evaluator.evaluate(upstage_text)
upstage_results['model'] = 'upstage_ocr'
upstage_results['file'] = basename
all_results.append(upstage_results)
except Exception:
# 오류가 있는 파일은 건너뜀
continue
df = pd.DataFrame(all_results)
# 컬럼 순서 재정렬
ordered_cols = ['file', 'model', 'strict_wer', 'strict_cer', 'flexible_wer', 'flexible_cer', 'word_hits', 'word_substitutions', 'word_deletions', 'word_insertions', 'char_hits', 'char_substitutions', 'char_deletions', 'char_insertions']
return df[ordered_cols]
# --- 콜백 함수 ---
def handle_nav_button(direction, total_files):
if direction == "prev" and st.session_state.eval_current_index > 0:
st.session_state.eval_current_index -= 1
elif direction == "next" and st.session_state.eval_current_index < total_files - 1:
st.session_state.eval_current_index += 1
def handle_selectbox_change():
st.session_state.eval_current_index = st.session_state.eval_selectbox_key
# --- 메인 UI 로직 ---
def main():
st.set_page_config(layout="wide", page_title="OCR 성능 평가 도구")
st.title("OCR 성능 평가 도구")
if "eval_current_index" not in st.session_state:
st.session_state.eval_current_index = 0
st.sidebar.header("세션 선택")
sessions = get_evaluable_sessions()
if not sessions:
st.info("평가 가능한 세션이 없습니다. 먼저 '정답셋 생성 도구'에서 정답셋을 생성해주세요.")
return
seed = st.sidebar.selectbox("평가할 세션을 선택하세요.", sessions)
if not seed:
st.info("사이드바에서 평가할 세션을 선택하세요.")
return
matched_files = match_evaluation_files(seed)
if matched_files is None:
st.error(f"'{seed}'에 해당하는 세션을 찾을 수 없거나, 필요한 폴더(docs, groundtruth 등)가 없습니다.")
return
if not matched_files:
st.warning("해당 세션에 평가할 파일(정답셋이 생성된 파일)이 없습니다.")
return
sorted_basenames = sorted(list(matched_files.keys()))
if st.session_state.eval_current_index >= len(sorted_basenames):
st.session_state.eval_current_index = 0
st.sidebar.header("파일 선택")
st.sidebar.selectbox(
"평가할 파일을 선택하세요.",
options=range(len(sorted_basenames)),
format_func=lambda x: f"{x+1}. {sorted_basenames[x]}",
index=st.session_state.eval_current_index,
key="eval_selectbox_key",
on_change=handle_selectbox_change,
)
st.sidebar.header("보기 옵션")
hide_document = st.sidebar.checkbox("원본 문서 숨기기", value=False)
st.sidebar.header("내보내기")
results_df = generate_all_results_df(matched_files)
if not results_df.empty:
csv = results_df.to_csv(index=False).encode('utf-8')
st.sidebar.download_button(
label="전체 결과 CSV 다운로드",
data=csv,
file_name=f"evaluation_results_{seed}.csv",
mime="text/csv",
)
else:
st.sidebar.write("다운로드할 결과가 없습니다.")
current_basename = sorted_basenames[st.session_state.eval_current_index]
nav_cols = st.columns([1, 5, 1])
nav_cols[0].button(
"◀ 이전",
on_click=handle_nav_button,
args=("prev", len(sorted_basenames)),
use_container_width=True,
)
nav_cols[1].markdown(
f"<h4 style='text-align: center; white-space: nowrap; overflow: hidden; text-overflow: ellipsis;'>{current_basename} ({st.session_state.eval_current_index + 1}/{len(sorted_basenames)})</h4>",
unsafe_allow_html=True,
)
nav_cols[2].button(
"다음 ▶",
on_click=handle_nav_button,
args=("next", len(sorted_basenames)),
use_container_width=True,
)
st.markdown("---")
files_to_evaluate = matched_files[current_basename]
if hide_document:
display_evaluation_for_file(files_to_evaluate)
else:
col1, col2 = st.columns([1, 1])
with col1:
st.header("📄 원본 문서")
doc_file = files_to_evaluate["doc_file"]
if doc_file.suffix.lower() == ".pdf":
display_pdf(doc_file)
else:
st.image(str(doc_file), use_container_width=True)
with col2:
display_evaluation_for_file(files_to_evaluate)
if __name__ == "__main__":
main()

92
workspace/ocr_eval_engine.py Normal file
View File

@@ -0,0 +1,92 @@
import jiwer
from fuzzywuzzy import fuzz
class OCREvaluator:
"""
정답(GT) 텍스트와 하나 이상의 예측(Hypothesis) 텍스트를 비교하여
다양한 문자 오류율(CER) 및 단어 오류율(WER) 지표를 계산하는 클래스.
"""
def __init__(self, ground_truth_text: str):
"""
평가기 인스턴스를 초기화합니다.
:param ground_truth_text: 비교의 기준이 되는 정답 텍스트.
"""
self.ground_truth = ground_truth_text
def evaluate(self, hypothesis_text: str) -> dict:
"""
주어진 예측 텍스트에 대한 모든 평가 지표를 계산합니다.
:param hypothesis_text: 평가할 OCR 예측 텍스트.
:return: 평가 결과를 담은 딕셔너리.
"""
cer_results = self._calculate_strict_cer(self.ground_truth, hypothesis_text)
wer_results = self._calculate_strict_wer(self.ground_truth, hypothesis_text)
flexible_cer = self._calculate_flexible_cer(self.ground_truth, hypothesis_text)
flexible_wer = self._calculate_flexible_wer(self.ground_truth, hypothesis_text)
results = {
# Strict CER
"strict_cer": cer_results["cer"],
"char_substitutions": cer_results["S"],
"char_deletions": cer_results["D"],
"char_insertions": cer_results["I"],
"char_hits": cer_results["H"],
# Strict WER
"strict_wer": wer_results["wer"],
"word_substitutions": wer_results["S"],
"word_deletions": wer_results["D"],
"word_insertions": wer_results["I"],
"word_hits": wer_results["H"],
# Flexible Metrics
"flexible_cer": flexible_cer,
"flexible_wer": flexible_wer,
}
return results
def _calculate_strict_cer(self, ref: str, hyp: str) -> dict:
"""
jiwer를 사용하여 엄격한 순서의 CER을 계산합니다.
"""
if not ref:
return {"cer": 1.0 if hyp else 0.0, "S": 0, "D": 0, "I": len(hyp), "H": 0}
output = jiwer.process_characters(ref, hyp)
return {
"cer": output.cer,
"S": output.substitutions,
"D": output.deletions,
"I": output.insertions,
"H": output.hits,
}
def _calculate_strict_wer(self, ref: str, hyp: str) -> dict:
"""
jiwer를 사용하여 엄격한 순서의 WER을 계산합니다.
"""
if not ref:
return {"wer": 1.0 if hyp else 0.0, "S": 0, "D": 0, "I": len(hyp.split()), "H": 0}
output = jiwer.process_words(ref, hyp)
return {
"wer": output.wer,
"S": output.substitutions,
"D": output.deletions,
"I": output.insertions,
"H": output.hits,
}
def _calculate_flexible_cer(self, ref: str, hyp: str) -> float:
"""
fuzzywuzzy의 token_sort_ratio를 사용하여 순서에 유연한 CER을 계산합니다.
"""
similarity_ratio = fuzz.token_sort_ratio(ref, hyp)
return (100 - similarity_ratio) / 100.0
def _calculate_flexible_wer(self, ref: str, hyp: str) -> float:
"""
fuzzywuzzy의 token_set_ratio를 사용하여 순서에 유연한 WER을 계산합니다.
"""
similarity_ratio = fuzz.token_set_ratio(ref, hyp)
return (100 - similarity_ratio) / 100.0