57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Dict, List
|
|
|
|
logger = logging.getLogger("line_manager")
|
|
|
|
class LineManager:
|
|
def __init__(self, config_file: str):
|
|
self.lines: Dict[str, dict] = {}
|
|
self.equipment_to_prefix: Dict[int, str] = {}
|
|
self.load_config(config_file)
|
|
|
|
def load_config(self, config_file: str):
|
|
try:
|
|
if os.path.exists(config_file):
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
|
|
for line_cfg in config.get('lines', []):
|
|
line_id = line_cfg.get('line_id')
|
|
equipment_id = line_cfg.get('equipment_id')
|
|
|
|
self.lines[line_id] = line_cfg
|
|
self.equipment_to_prefix[equipment_id] = line_id
|
|
|
|
logger.info(f"[LINE CONFIG] Loaded {len(self.lines)} line configurations")
|
|
else:
|
|
logger.warning(f"[LINE CONFIG] Config file not found: {config_file}. Defaults to line1 -> ID 8.")
|
|
except Exception as e:
|
|
logger.error(f"[LINE CONFIG ERROR] Failed to load config: {e}")
|
|
|
|
def get_equipment_id(self, prefix: str) -> int:
|
|
if prefix in self.lines:
|
|
return self.lines[prefix].get('equipment_id', 8)
|
|
# Default fallback
|
|
if "line1" in prefix:
|
|
return 8
|
|
match = re.search(r'\d+', prefix)
|
|
if match:
|
|
line_num = int(match.group())
|
|
return 7 + line_num
|
|
return 8
|
|
|
|
def get_prefix(self, equipment_id: int) -> str:
|
|
return self.equipment_to_prefix.get(equipment_id, "line1")
|
|
|
|
def get_all_prefixes(self) -> List[str]:
|
|
if not self.lines:
|
|
return ["line1"]
|
|
return list(self.lines.keys())
|
|
|
|
# 전역 인스턴스
|
|
config_path = os.path.join(os.path.dirname(__file__), "line_config.json")
|
|
line_manager = LineManager(config_path)
|