intelligent-pid / utils.py
msIntui
Initial commit: Add core files for P&ID processing
9847531
Raw
History Blame Contribute Delete
4.05 kB
import time
import numpy as np
from contextlib import contextmanager
from loguru import logger
from typing import List, Dict, Optional, Tuple, Union
from detection_schema import BBox
from storage import StorageInterface
class DebugHandler:
"""Production-grade debugging and performance tracking"""
def __init__(self, enabled: bool = False, storage: StorageInterface = None):
self.enabled = enabled
self.storage = storage
self.metrics = {}
self._start_time = None
@contextmanager
def track_performance(self, operation_name: str):
"""Context manager for performance tracking"""
if self.enabled:
self._start_time = time.perf_counter()
logger.debug(f"Starting {operation_name}")
yield
if self.enabled:
duration = time.perf_counter() - self._start_time
self.metrics[operation_name] = duration
logger.debug(f"{operation_name} completed in {duration:.2f}s")
def save_artifact(self, name: str, data: bytes, extension: str = "png"):
"""Generic artifact storage handler"""
if self.enabled and self.storage:
path = f"debug/{name}.{extension}"
self.storage.save_file(path, data)
logger.info(f"Saved debug artifact: {path}")
class CoordinateTransformer:
@staticmethod
def global_to_local_bbox(
bbox: Union[BBox, List[BBox]],
roi: Optional[np.ndarray]
) -> Union[BBox, List[BBox]]:
"""
Convert global BBox(es) to ROI-local coordinates
Handles both single BBox and lists of BBoxes
"""
if roi is None or len(roi) != 4:
return bbox
x_min, y_min, _, _ = roi
def convert(b: BBox) -> BBox:
return BBox(
xmin=b.xmin - x_min,
ymin=b.ymin - y_min,
xmax=b.xmax - x_min,
ymax=b.ymax - y_min
)
return map(convert, bbox) if isinstance(bbox, list) else convert(bbox)
@staticmethod
def local_to_global_bbox(
bbox: Union[BBox, List[BBox]],
roi: Optional[np.ndarray]
) -> Union[BBox, List[BBox]]:
"""
Convert ROI-local BBox(es) to global coordinates
Handles both single BBox and lists of BBoxes
"""
if roi is None or len(roi) != 4:
return bbox
x_min, y_min, _, _ = roi
def convert(b: BBox) -> BBox:
return BBox(
xmin=b.xmin + x_min,
ymin=b.ymin + y_min,
xmax=b.xmax + x_min,
ymax=b.ymax + y_min
)
return map(convert, bbox) if isinstance(bbox, list) else convert(bbox)
# Maintain legacy tuple support if needed
@staticmethod
def global_to_local(
bboxes: List[Tuple[int, int, int, int]],
roi: Optional[np.ndarray]
) -> List[Tuple[int, int, int, int]]:
"""Legacy tuple version for backward compatibility"""
if roi is None or len(roi) != 4:
return bboxes
x_min, y_min, _, _ = roi
return [(x1 - x_min, y1 - y_min, x2 - x_min, y2 - y_min)
for x1, y1, x2, y2 in bboxes]
@staticmethod
def local_to_global(
bboxes: List[Tuple[int, int, int, int]],
roi: Optional[np.ndarray]
) -> List[Tuple[int, int, int, int]]:
"""Legacy tuple version for backward compatibility"""
if roi is None or len(roi) != 4:
return bboxes
x_min, y_min, _, _ = roi
return [(x1 + x_min, y1 + y_min, x2 + x_min, y2 + y_min)
for x1, y1, x2, y2 in bboxes]
@staticmethod
def local_to_global_point(point: Tuple[int, int], roi: Optional[np.ndarray]) -> Tuple[int, int]:
"""Convert single point from local to global coordinates"""
if roi is None or len(roi) != 4:
return point
x_min, y_min, _, _ = roi
return (int(point[0] + x_min), int(point[1] + y_min))