Initial commit: HC900 Crawler
Honeywell HC900을 Modbus TCP로 직접 폴링 → gRPC → C# 크롤러 → PostgreSQL. 기존 Experion OPC UA 데이터 경로를 HC900 직접 통신으로 대체. - industrial-comm/cpp: C++ Modbus 게이트웨이 (gRPC 서버) - src: C# .NET 8 ASP.NET Core 크롤러 + 웹 UI (3-Layer) - mcp-server: Python FastMCP (RAG/NL2SQL/P&ID) - 다중 컨트롤러(N-Controller) 지원 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
windpacer
1
mcp-server/instrument_inference/__init__.py
Normal file
1
mcp-server/instrument_inference/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Field instrument inference from DCS base_tag to field instruments."""
|
||||
171
mcp-server/instrument_inference/excel.py
Normal file
171
mcp-server/instrument_inference/excel.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""3시트 Excel 초안 생성기 (§3 스키마)."""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
|
||||
import openpyxl
|
||||
from openpyxl.styles import Font, PatternFill, Alignment
|
||||
|
||||
from .rules import get_all_measurements, get_all_modifiers, get_all_special_prefixes
|
||||
|
||||
|
||||
INSTRUMENT_COLS = [
|
||||
"instrument_id", "display_name", "parent_base_tag", "role", "loop",
|
||||
"area", "measures", "data_points", "from", "to", "description",
|
||||
"confidence", "needs_review", "inference_basis", "operator_notes", "delete",
|
||||
]
|
||||
|
||||
POWER_EQUIPMENT_COLS = [
|
||||
"instrument_id", "display_name", "parent_base_tag", "role", "loop",
|
||||
"area", "equipment_type", "data_points", "description",
|
||||
"confidence", "needs_review", "inference_basis", "operator_notes", "delete",
|
||||
]
|
||||
|
||||
HEADER_FILL = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
|
||||
HEADER_FONT = Font(bold=True, color="FFFFFF", size=11)
|
||||
LOW_FILL = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
|
||||
MEDIUM_FILL = PatternFill(start_color="FFEB9C", end_color="FFEB9C", fill_type="solid")
|
||||
|
||||
|
||||
def generate_excel(instruments: list[dict], unmatched: list[str], power_equipment: list[dict] | None = None) -> str:
|
||||
"""
|
||||
4시트 Excel 생성 후 파일 경로 반환.
|
||||
|
||||
Args:
|
||||
instruments: infer.py 결과 (instrument dict 리스트)
|
||||
unmatched: 룰 미매칭 base_tag 목록
|
||||
power_equipment: 동력기기 (펌프, 압축기, 교반기 등)
|
||||
output_dir: 출력 디렉토리 (기본: /tmp)
|
||||
|
||||
Returns:
|
||||
생성된 xlsx 파일의 절대 경로
|
||||
"""
|
||||
output_dir = tempfile.gettempdir()
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
filepath = os.path.join(output_dir, f"instruments_draft_{ts}.xlsx")
|
||||
|
||||
wb = openpyxl.Workbook()
|
||||
|
||||
# ── 시트 1: instruments ──
|
||||
_build_instruments_sheet(wb, instruments)
|
||||
|
||||
# ── 시트 2: power_equipment ──
|
||||
_build_power_equipment_sheet(wb, power_equipment or [])
|
||||
|
||||
# ── 시트 3: unmatched_tags ──
|
||||
_build_unmatched_sheet(wb, unmatched)
|
||||
|
||||
# ── 시트 4: naming_convention ──
|
||||
_build_naming_convention_sheet(wb)
|
||||
|
||||
wb.save(filepath)
|
||||
return filepath
|
||||
|
||||
|
||||
def _build_instruments_sheet(wb: openpyxl.Workbook, instruments: list[dict]) -> None:
|
||||
ws = wb.active
|
||||
ws.title = "instruments"
|
||||
|
||||
# 헤더
|
||||
for col_idx, col_name in enumerate(INSTRUMENT_COLS, 1):
|
||||
cell = ws.cell(row=1, column=col_idx, value=col_name)
|
||||
cell.font = HEADER_FONT
|
||||
cell.fill = HEADER_FILL
|
||||
cell.alignment = Alignment(horizontal="center", wrap_text=True)
|
||||
|
||||
# 데이터 행
|
||||
for row_idx, inst in enumerate(instruments, 2):
|
||||
for col_idx, col_name in enumerate(INSTRUMENT_COLS, 1):
|
||||
val = inst.get(col_name, "")
|
||||
cell = ws.cell(row=row_idx, column=col_idx, value=val)
|
||||
|
||||
# confidence별 색상
|
||||
if col_name == "confidence":
|
||||
if val == "low":
|
||||
cell.fill = LOW_FILL
|
||||
elif val == "medium":
|
||||
cell.fill = MEDIUM_FILL
|
||||
|
||||
# 컬럼 너비 자동 조정
|
||||
for col_idx in range(1, len(INSTRUMENT_COLS) + 1):
|
||||
max_len = 0
|
||||
for row in range(1, ws.max_row + 1):
|
||||
val = str(ws.cell(row=row, column=col_idx).value or "")
|
||||
max_len = max(max_len, len(val))
|
||||
ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = min(max_len + 4, 40)
|
||||
|
||||
|
||||
def _build_naming_convention_sheet(wb: openpyxl.Workbook) -> None:
|
||||
ws = wb.create_sheet("naming_convention")
|
||||
|
||||
# 측정량 표
|
||||
ws.cell(row=1, column=1, value="측정량 (첫 글자)").font = HEADER_FONT
|
||||
ws.cell(row=1, column=2, value="의미").font = HEADER_FONT
|
||||
meas = get_all_measurements()
|
||||
for i, (letter, meaning) in enumerate(meas.items(), 2):
|
||||
ws.cell(row=i, column=1, value=letter)
|
||||
ws.cell(row=i, column=2, value=meaning)
|
||||
|
||||
# 수식어 표 (D열부터)
|
||||
ws.cell(row=1, column=4, value="수식어 (두 번째 이후)").font = HEADER_FONT
|
||||
ws.cell(row=1, column=5, value="role").font = HEADER_FONT
|
||||
ws.cell(row=1, column=6, value="가상").font = HEADER_FONT
|
||||
mods = get_all_modifiers()
|
||||
for i, (letter, info) in enumerate(mods.items(), 2):
|
||||
ws.cell(row=i, column=4, value=letter)
|
||||
ws.cell(row=i, column=5, value=info.get("role", ""))
|
||||
ws.cell(row=i, column=6, value=info.get("virtual", False))
|
||||
|
||||
# 특수 prefix 표 (G열부터)
|
||||
ws.cell(row=1, column=7, value="특수 prefix").font = HEADER_FONT
|
||||
ws.cell(row=1, column=8, value="role").font = HEADER_FONT
|
||||
sps = get_all_special_prefixes()
|
||||
for i, (prefix, info) in enumerate(sps.items(), 2):
|
||||
ws.cell(row=i, column=7, value=prefix)
|
||||
ws.cell(row=i, column=8, value=info.get("role", ""))
|
||||
|
||||
|
||||
def _build_unmatched_sheet(wb: openpyxl.Workbook, unmatched: list[str]) -> None:
|
||||
ws = wb.create_sheet("unmatched_tags")
|
||||
|
||||
ws.cell(row=1, column=1, value="base_tag").font = HEADER_FONT
|
||||
ws.cell(row=1, column=2, value="area").font = HEADER_FONT
|
||||
ws.cell(row=1, column=3, value="action").font = HEADER_FONT
|
||||
ws.cell(row=1, column=4, value="operator_notes").font = HEADER_FONT
|
||||
|
||||
for i, tag in enumerate(unmatched, 2):
|
||||
ws.cell(row=i, column=1, value=tag)
|
||||
ws.cell(row=i, column=2, value="(none)")
|
||||
ws.cell(row=i, column=3, value="운영자가 instruments 시트에 행 추가 필요")
|
||||
ws.cell(row=i, column=4, value="")
|
||||
|
||||
|
||||
def _build_power_equipment_sheet(wb: openpyxl.Workbook, equipment: list[dict]) -> None:
|
||||
ws = wb.create_sheet("power_equipment")
|
||||
|
||||
for col_idx, col_name in enumerate(POWER_EQUIPMENT_COLS, 1):
|
||||
cell = ws.cell(row=1, column=col_idx, value=col_name)
|
||||
cell.font = HEADER_FONT
|
||||
cell.fill = HEADER_FILL
|
||||
cell.alignment = Alignment(horizontal="center", wrap_text=True)
|
||||
|
||||
for row_idx, eq in enumerate(equipment, 2):
|
||||
for col_idx, col_name in enumerate(POWER_EQUIPMENT_COLS, 1):
|
||||
val = eq.get(col_name, "")
|
||||
cell = ws.cell(row=row_idx, column=col_idx, value=val)
|
||||
|
||||
if col_name == "confidence":
|
||||
if val == "low":
|
||||
cell.fill = LOW_FILL
|
||||
elif val == "medium":
|
||||
cell.fill = MEDIUM_FILL
|
||||
|
||||
for col_idx in range(1, len(POWER_EQUIPMENT_COLS) + 1):
|
||||
max_len = 0
|
||||
for row in range(1, ws.max_row + 1):
|
||||
val = str(ws.cell(row=row, column=col_idx).value or "")
|
||||
max_len = max(max_len, len(val))
|
||||
ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = min(max_len + 4, 40)
|
||||
329
mcp-server/instrument_inference/infer.py
Normal file
329
mcp-server/instrument_inference/infer.py
Normal file
@@ -0,0 +1,329 @@
|
||||
"""DCS base_tag → 현장 계기 자동 유추 알고리즘."""
|
||||
from __future__ import annotations
|
||||
import re
|
||||
|
||||
from .rules import get_measurement, get_modifier, get_special_prefix
|
||||
|
||||
|
||||
def split_tag(base_tag: str) -> tuple[str, str]:
|
||||
"""'ficq-6101' → ('ficq', '6101') / 'xv-6124' → ('xv', '6124')."""
|
||||
m = re.match(r"^([a-zA-Z]+)-?(.*)", base_tag)
|
||||
if not m:
|
||||
return base_tag, ""
|
||||
return m.group(1).lower(), m.group(2)
|
||||
|
||||
|
||||
def infer_instruments_for_base_tag(
|
||||
base_tag: str,
|
||||
data_points: list[str],
|
||||
area: str,
|
||||
) -> list[dict]:
|
||||
"""
|
||||
단일 base_tag에 대해 현장 계기 목록을 유추.
|
||||
|
||||
Returns:
|
||||
instrument dict 리스트 (§3.1 스키마와 일치).
|
||||
매칭 실패 시 1행 포함 (confidence=low, needs_review=True).
|
||||
"""
|
||||
head, loop = split_tag(base_tag)
|
||||
if not loop:
|
||||
loop = "unknown"
|
||||
|
||||
dp_set = set(data_points) if data_points else set()
|
||||
|
||||
# 규칙에서 어긋난 태그 → unmatched
|
||||
# loop에 '_' 포함 (fica-3102_op 등) 또는 'esd' 포함 (lt-9113-lo-esd 등 시스템 포인트)
|
||||
if "_" in loop or "esd" in loop.lower():
|
||||
return _build_unmatched(base_tag, area, head, loop)
|
||||
|
||||
# 1. 특수 prefix 우선 (정확한 전체 일치)
|
||||
sp = get_special_prefix(head)
|
||||
if sp:
|
||||
return _build_special(head, loop, area, sp, dp_set)
|
||||
|
||||
# 2. 첫 글자 = 측정량
|
||||
first = head[0]
|
||||
meas = get_measurement(first)
|
||||
if not meas:
|
||||
return _build_unmatched(base_tag, area, head, loop)
|
||||
|
||||
instruments = []
|
||||
has_transmitter = False
|
||||
has_controller = False
|
||||
|
||||
# DCS 내부 기능: controller, totalizer, alarm, switch는 현장 계기가 아님
|
||||
_dcs_internal_roles = {"controller", "totalizer", "alarm", "switch"}
|
||||
|
||||
# 3. 수식어 글자별로 계기 생성
|
||||
for letter in head[1:]:
|
||||
mod = get_modifier(letter)
|
||||
if not mod:
|
||||
continue
|
||||
if mod.get("virtual"):
|
||||
continue # I, R은 가상
|
||||
|
||||
role = mod["role"]
|
||||
|
||||
# DCS 내부 기능은 instruments에서 제외
|
||||
if role in _dcs_internal_roles:
|
||||
if role == "controller":
|
||||
has_controller = True
|
||||
continue
|
||||
|
||||
inst = _build_instrument(first, meas, role, loop, area, mod, dp_set, base_tag)
|
||||
instruments.append(inst)
|
||||
|
||||
if role == "transmitter":
|
||||
has_transmitter = True
|
||||
|
||||
# 4. T 글자가 명시 안 됐어도 컨트롤러가 있으면 송신기 암시
|
||||
if not has_transmitter and has_controller:
|
||||
inst = _build_implicit_transmitter(first, meas, loop, area, dp_set, base_tag)
|
||||
instruments.insert(0, inst)
|
||||
has_transmitter = True
|
||||
|
||||
# 5. 컨트롤러 → 제어밸브 자동 생성 (auto_pair)
|
||||
if has_controller:
|
||||
inst = _build_paired_valve(first, meas, loop, area, dp_set, base_tag)
|
||||
instruments.append(inst)
|
||||
|
||||
# [H-4 수정] 빈 리스트일 때 implicit transmitter 강제 추가
|
||||
# pi-XXXX/ti-XXXX 등 "측정량 + I(virtual)"만 있는 경우 instruments가 비어 있음
|
||||
if not instruments:
|
||||
inst = _build_implicit_transmitter(first, meas, loop, area, dp_set, base_tag)
|
||||
instruments.append(inst)
|
||||
|
||||
# 6. from/to 채우기
|
||||
_link_signal_flow(instruments, base_tag, area)
|
||||
|
||||
# 7. confidence 계산
|
||||
for inst in instruments:
|
||||
inst["confidence"] = _score_confidence(inst, dp_set)
|
||||
inst["needs_review"] = inst["confidence"] == "low"
|
||||
|
||||
return instruments
|
||||
|
||||
|
||||
def _build_instrument(
|
||||
first_letter: str, meas: str, role: str, loop: str, area: str, mod: dict, dp_set: set, parent_tag: str
|
||||
) -> dict:
|
||||
role_id = _role_to_id(first_letter, role, loop)
|
||||
display = f"{first_letter.upper()}{_role_suffix(role).upper()}-{loop.upper()}"
|
||||
dps = mod.get("data_points", [])
|
||||
matched_dps = [d for d in dps if d in dp_set]
|
||||
|
||||
return {
|
||||
"instrument_id": role_id,
|
||||
"display_name": display,
|
||||
"parent_base_tag": parent_tag,
|
||||
"role": role,
|
||||
"loop": loop,
|
||||
"area": area or "(none)",
|
||||
"measures": meas,
|
||||
"data_points": ",".join(matched_dps) if matched_dps else "(none)",
|
||||
"from": "(none)",
|
||||
"to": "(none)",
|
||||
"description": "",
|
||||
"confidence": "medium",
|
||||
"needs_review": False,
|
||||
"inference_basis": f"{first_letter.upper()}+{mod.get('role','?')}",
|
||||
"operator_notes": "",
|
||||
"delete": False,
|
||||
}
|
||||
|
||||
|
||||
def _build_special(head: str, loop: str, area: str, sp: dict, dp_set: set) -> list[dict]:
|
||||
role = sp["role"]
|
||||
meas = sp.get("measures")
|
||||
equip_type = sp.get("equipment_type")
|
||||
role_id = f"{head}-{loop}"
|
||||
display = f"{head.upper()}-{loop.upper()}"
|
||||
|
||||
dp_map = {
|
||||
"shutdown-valve": [".instate0", ".instate1"],
|
||||
"interlock-relay": [".instate0", ".instate1"],
|
||||
"positioner": [".op"],
|
||||
"power_equipment": [".run", ".fault"],
|
||||
}
|
||||
expected_dps = dp_map.get(role, [])
|
||||
matched_dps = [d for d in expected_dps if d in dp_set]
|
||||
|
||||
inst = {
|
||||
"instrument_id": role_id,
|
||||
"display_name": display,
|
||||
"parent_base_tag": f"{head}-{loop}",
|
||||
"role": role,
|
||||
"loop": loop,
|
||||
"area": area or "(none)",
|
||||
"measures": meas or "(none)",
|
||||
"data_points": ",".join(matched_dps) if matched_dps else "(none)",
|
||||
"from": "(none)",
|
||||
"to": "(none)",
|
||||
"description": "",
|
||||
"confidence": "medium",
|
||||
"needs_review": False,
|
||||
"inference_basis": f"special_prefix:{head}",
|
||||
"operator_notes": "",
|
||||
"delete": False,
|
||||
}
|
||||
|
||||
if role == "power_equipment" and equip_type:
|
||||
inst["equipment_type"] = equip_type
|
||||
return [inst]
|
||||
|
||||
|
||||
def _build_unmatched(base_tag: str, area: str, head: str, loop: str) -> list[dict]:
|
||||
return [{
|
||||
"instrument_id": f"{head}-{loop}",
|
||||
"display_name": f"{head.upper()}-{loop.upper()}",
|
||||
"parent_base_tag": base_tag,
|
||||
"role": "equipment",
|
||||
"loop": loop,
|
||||
"area": area or "(none)",
|
||||
"measures": "(none)",
|
||||
"data_points": "(none)",
|
||||
"from": "(none)",
|
||||
"to": "(none)",
|
||||
"description": "",
|
||||
"confidence": "low",
|
||||
"needs_review": True,
|
||||
"inference_basis": "unmatched_prefix",
|
||||
"operator_notes": "",
|
||||
"delete": False,
|
||||
}]
|
||||
|
||||
|
||||
def _build_implicit_transmitter(first: str, meas: str, loop: str, area: str, dp_set: set, parent: str) -> dict:
|
||||
role_id = f"{first}t-{loop}"
|
||||
return {
|
||||
"instrument_id": role_id,
|
||||
"display_name": f"{first.upper()}T-{loop.upper()}",
|
||||
"parent_base_tag": parent,
|
||||
"role": "transmitter",
|
||||
"loop": loop,
|
||||
"area": area or "(none)",
|
||||
"measures": meas,
|
||||
"data_points": ".pv" if ".pv" in dp_set else "(none)",
|
||||
"from": f"process/{area or 'unknown'}-{loop}-inlet",
|
||||
"to": f"tag/{parent}",
|
||||
"description": "",
|
||||
"confidence": "medium",
|
||||
"needs_review": False,
|
||||
"inference_basis": f"{first.upper()}+(implied T)",
|
||||
"operator_notes": "",
|
||||
"delete": False,
|
||||
}
|
||||
|
||||
|
||||
def _build_paired_valve(first: str, meas: str, loop: str, area: str, dp_set: set, parent: str) -> dict:
|
||||
role_id = f"{first}cv-{loop}"
|
||||
return {
|
||||
"instrument_id": role_id,
|
||||
"display_name": f"{first.upper()}CV-{loop.upper()}",
|
||||
"parent_base_tag": parent,
|
||||
"role": "control-valve",
|
||||
"loop": loop,
|
||||
"area": area or "(none)",
|
||||
"measures": "(none)",
|
||||
"data_points": ".op" if ".op" in dp_set else "(none)",
|
||||
"from": f"tag/{first}ic-{loop}",
|
||||
"to": f"process/{area or 'unknown'}-{loop}-downstream",
|
||||
"description": "",
|
||||
"confidence": "high",
|
||||
"needs_review": False,
|
||||
"inference_basis": "C -> CV auto_pair",
|
||||
"operator_notes": "",
|
||||
"delete": False,
|
||||
}
|
||||
|
||||
|
||||
def _role_suffix(role: str) -> str:
|
||||
mapping = {
|
||||
"transmitter": "t",
|
||||
"controller": "ic",
|
||||
"totalizer": "q",
|
||||
"switch": "s",
|
||||
"alarm": "a",
|
||||
"interlock-relay": "y",
|
||||
"positioner": "z",
|
||||
}
|
||||
return mapping.get(role, role[:2])
|
||||
|
||||
|
||||
def _role_to_id(first_letter: str, role: str, loop: str) -> str:
|
||||
return f"{first_letter}{_role_suffix(role)}-{loop}"
|
||||
|
||||
|
||||
def _link_signal_flow(instruments: list[dict], parent_tag: str, area: str) -> None:
|
||||
"""role별 from/to 기본값 채음 (§5.1)."""
|
||||
transmitters = [i for i in instruments if i["role"] == "transmitter"]
|
||||
valves = [i for i in instruments if i["role"] == "control-valve"]
|
||||
|
||||
for t in transmitters:
|
||||
if t["from"] == "(none)":
|
||||
t["from"] = f"process/{area or 'unknown'}-{t['loop']}-inlet"
|
||||
if t["to"] == "(none)":
|
||||
t["to"] = f"tag/{parent_tag}"
|
||||
|
||||
for v in valves:
|
||||
if v["from"] == "(none)":
|
||||
v["from"] = f"tag/{parent_tag}"
|
||||
if v["to"] == "(none)":
|
||||
v["to"] = f"process/{area or 'unknown'}-{v['loop']}-downstream"
|
||||
|
||||
|
||||
def _score_confidence(inst: dict, dp_set: set) -> str:
|
||||
"""data_point 일치도 + prefix 매칭으로 신뢰도 계산."""
|
||||
role = inst["role"]
|
||||
basis = inst.get("inference_basis", "")
|
||||
|
||||
if basis == "unmatched_prefix":
|
||||
return "low"
|
||||
|
||||
expected = {
|
||||
"transmitter": [".pv"],
|
||||
"controller": [".sp", ".op"],
|
||||
"totalizer": [".qv"],
|
||||
"switch": [".instate0"],
|
||||
"shutdown-valve": [".instate0"],
|
||||
}
|
||||
|
||||
checks = expected.get(role, [])
|
||||
if not checks:
|
||||
return "medium" if basis.startswith("special") else "high"
|
||||
|
||||
matched = sum(1 for c in checks if c in dp_set)
|
||||
if matched == len(checks):
|
||||
return "high"
|
||||
if matched > 0:
|
||||
return "medium"
|
||||
return "low"
|
||||
|
||||
|
||||
def _role_to_korean_description(role: str, meas: str) -> str:
|
||||
"""role + 측정량 → 한국어 설명 초안."""
|
||||
meas_ko = {
|
||||
"flow": "유량", "pressure": "압력", "temperature": "온도",
|
||||
"level": "위차", "analysis": "분석", "speed": "회전수",
|
||||
"weight": "중량", "density": "비중", "power": "전력", "moisture": "함량",
|
||||
}
|
||||
m = meas_ko.get(meas, meas)
|
||||
|
||||
role_desc = {
|
||||
"transmitter": f"{m} 송신기",
|
||||
"controller": f"{m} 제어기",
|
||||
"totalizer": f"{m} 적산기",
|
||||
"switch": f"{m} 스위치",
|
||||
"alarm": f"{m} 알람",
|
||||
"control-valve": f"{m} 제어밸브",
|
||||
"shutdown-valve": "차단밸브",
|
||||
"interlock-relay": "인터록 릴레이",
|
||||
"positioner": "포지셔너",
|
||||
"motor": "모터",
|
||||
"pump": "펌프",
|
||||
"compressor": "압축기",
|
||||
"agitator": "교반기",
|
||||
"blower": "송풍기",
|
||||
"fan": "송풍기",
|
||||
}
|
||||
return role_desc.get(role, role)
|
||||
47
mcp-server/instrument_inference/rules.py
Normal file
47
mcp-server/instrument_inference/rules.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""YAML 룰 로더 — prompts/instrument_inference.yaml을 읽고 캐싱."""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
import yaml
|
||||
from functools import lru_cache
|
||||
|
||||
_RULES_PATH = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"..", "..", "prompts", "instrument_inference.yaml",
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def load_rules() -> dict:
|
||||
"""YAML 룰 파일 로드 (lru_cache로 1회만 로드)."""
|
||||
with open(_RULES_PATH, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
|
||||
def get_measurement(letter: str) -> str | None:
|
||||
"""첫 글자 → 측정량 반환."""
|
||||
return load_rules().get("measurement", {}).get(letter)
|
||||
|
||||
|
||||
def get_modifier(letter: str) -> dict | None:
|
||||
"""수식어 글자 → role 정보 반환."""
|
||||
return load_rules().get("modifiers", {}).get(letter)
|
||||
|
||||
|
||||
def get_special_prefix(head: str) -> dict | None:
|
||||
"""특수 prefix → role 정보 반환."""
|
||||
return load_rules().get("special_prefixes", {}).get(head)
|
||||
|
||||
|
||||
def get_all_measurements() -> dict:
|
||||
"""측정량 전체 표 반환 (naming_convention 시트용)."""
|
||||
return load_rules().get("measurement", {})
|
||||
|
||||
|
||||
def get_all_modifiers() -> dict:
|
||||
"""수식어 전체 표 반환 (naming_convention 시트용)."""
|
||||
return load_rules().get("modifiers", {})
|
||||
|
||||
|
||||
def get_all_special_prefixes() -> dict:
|
||||
"""특수 prefix 전체 표 반환 (naming_convention 시트용)."""
|
||||
return load_rules().get("special_prefixes", {})
|
||||
Reference in New Issue
Block a user