Inference

cycle_counter.CycleCounterInferenceHandler

학습한 Video Classification 모델로 프레임 단위 추론/사이클 추출을 수행하는 API 래퍼입니다.

참고

공개 메서드는 @error_handler로 감싸져 호출 시 (error_code: int, message: str, payload: Any) 형태로 반환됩니다. 각 메서드의 Returns 설명은 기본적으로 payload를 기준으로 작성되어 있습니다.

Usage

demo/demo_inference.py 참고

---

build(config) classmethod

InferenceHandler 인스턴스를 생성합니다.

Parameters:

Name	Type	Description	Default
config	dict	InferenceHandler를 build 하기 위한 config가 담긴 dictionary 입니다. { "model_name": str, # 모델 이름 "checkpoint_path": str, # 체크포인트 경로 "password": Optional[str], # 체크포인트 패스워드 "inference_options": Optional[Dict[str, Any]], # 추론 옵션 "device": Union[int, str], # GPU 번호 (int) 또는 "cpu" (str) (기본값 "cpu") "anomaly_by_cycle_detector": Optional[dict], # 주기 길이 이상치 검출 모델 설정 (미지정시 학습 데이터에서 추출된 mean/sigma 기반으로 z-test 자동 빌드) ```python { "mode": str, # 모드 (z-test, full_interval, marginal_interval). 기본값: "z-test" "test_mode": str, # 검정 방향 ("two-sided", "right-sided", "left-sided"). 기본값: "two-sided" "sigma_level_1": float, # level-1 임계 z-값 (95%). 기본값: 1.96 "sigma_level_2": float, # level-2 임계 z-값 (99%). 기본값: 2.576 "stats": Optional[dict], # z-test 통계 정보 # {class_id: {mean: float, sigma: float, test_mode?: str, sigma_level_1?: float, sigma_level_2?: float}, ...} # test_mode, sigma_level_1, sigma_level_2는 클래스별 오버라이드 (미지정 시 전역 설정 사용) "intervals": Optional[dict], # interval 모드 구간 정보 {class_id: {level_1_lower, level_1_upper, level_2_lower, level_2_upper}, ...} } ``` "frame_selector": Optional[dict], # 프레임 선택 모델 설정 ```python { "cycle_select_offset_by_class": dict[str, list[int]], ```python { "class_id": int, "offset": list[int] } ``` } ``` "detection": Optional[dict], # detection 엔진 설정 ```python { "checkpoint_path": str, # 체크포인트 경로 "password": Optional[str], # 체크포인트 패스워드 "device": Union[int, str], # GPU 번호 또는 "cpu" "inference_options": Optional[dict], # detection 엔진 추론 옵션 } ``` "det_template": Optional[dict | str], # detection tracking 템플릿(dict 또는 파일 경로) ```python { "waypoints": dict[str, list[tuple[int, int]]], # 템플릿 타입 (polygon, polyline) "phase_group_to_class": dict[str, int], # [(class_name: class_index)] "class_to_phases": dict[int, list[str]], # {class_index: [class_name,...]} } ``` "tracked_anomaly_threshold": float, # 경로 편차 이상치 판단 임계값 (기본값 5000) "use_ts": bool, # class segmentation에서 ms 기반의 ts 단위 사용 여부 (기본값 True) }	required

Returns:

Name	Type	Description
CycleCounterInferenceHandler	CycleCounterInferenceHandler	build가 완료된 API 래퍼 인스턴스를 반환합니다.

---

infer(data, frame_timestamps)

주어진 데이터를 사용하여 추론을 수행합니다.

Parameters:

Name	Type	Description	Default
data	List[ndarray]	Inference에 사용할 데이터 입니다. 데이터는 (H, W, C) 형태와 RGB 채널의 numpy array 입니다.	required
frame_timestamps	list[Union[int, float]]]	추론할 데이터에 대한 타임스탬프.	required

Returns:

Name	Type	Description
	InferenceStatus	tuple[int, str, tuple[InferenceStatus, list[InferenceOutput]]]: @error_handler 래핑 후의 실제 반환 형식입니다. payload는 (inference_status, inference_output) 입니다.
inference_status	InferenceStatus	추론 상태를 의미합니다. InferenceStatus.SELECTED인 경우 get_cycles()로 사이클을 조회할 수 있습니다.
inference_output	list[InferenceOutput]	각 프레임에 대한 추론 결과입니다. 각 요소는 다음 키를 가집니다. - score (float) - class_index (int) - error (int) - cams (Optional[np.ndarray]) - probs (Optional[float]) - logit (Optional[float]) - detection (Optional[dict]): detection 엔진 사용 시, 해당 프레임 timestamp에 매칭된 detection 결과. detection을 사용하지 않으면 None 입니다. { "detected_objects": [{"bounding_box": [l,t,w,h], "class_index": int, "score": int}], }

InferenceStatus

• INSPECT: 추론을 진행중이며, SELECT 과정에 있지 않습니다.
• SELECTING: 현재 검사대상 구간이 존재하며, 검사후보를 선택 중입니다.
• SELECTED: 검사후보를 선택했습니다.
• ERROR: 추론 중 에러가 발생했습니다.

InferenceOutput

• score (float): 예측된 class_index의 softmax score입니다. (0.0 ~ 1.0)
• class_index (int): 프레임별 예측 클래스 인덱스입니다.
  • energy_ood_enabled=True이고 energy score가 energy_ood_threshold를 초과하면 energy_ood_fallback_class_index로 class_index가 대체됩니다.
• error (int): 에러 코드입니다. (0 이면 정상입니다.)
• logit (Optional[float]): 예측된 class_index에 해당하는 softmax 이전의 원본 logit 값입니다.

Note

• 실제 emit/lookahead 정책은 현재 inference option 설정값을 따릅니다. (set_inference_option("emit_mode", ...), set_inference_option("lookahead", ...))

Usage

error, message, (inference_status, inference_output) = inference_handler.infer(
    data=sequence,
    frame_timestamps=timestamps,
)

if inference_status == InferenceStatus.SELECTED:
    error, message, cycles = inference_handler.get_cycles()

# infer API 내부에서 예외가 발생해도 API 레벨 error/message는 성공으로 반환될 수 있습니다.
# 대신 inference_status가 InferenceStatus.ERROR로 설정되며, error code는 inference_output에서 확인할 수 있습니다.
# inference_output 각 요소의 값은 0.0, 0, error_code로 설정됩니다.
if inference_status == InferenceStatus.ERROR:
    error_code = inference_output[0]["error"]

---

flush_pending_inference()

지연 emit 모드에서 남은 미방출 추론 결과를 강제로 방출합니다.

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: 각 항목은 다음 키를 포함합니다. - status (InferenceStatus) - outputs (list[InferenceOutput]) - emitted_timestamps (list[int | float])

Note

• 공식 지원은 emit_mode="previous" 입니다.
• emit_mode="current" 인 경우 빈 리스트를 반환합니다.

get_cycles()

현재까지 확정된 사이클 목록을 반환합니다.

Returns:

Type	Description
list[CycleInfo]	tuple[int, str, list[CycleInfo]]: @error_handler 래핑 후의 실제 반환 형식입니다.

주요 CycleInfo 키: - cycle_length: int - start_frame_ts, end_frame_ts: int | float - class_index: int - original_class_index: Optional[int] - current_model: str ("cls" 또는 "det" 등) - selected_frames: list[dict] - detections: Optional[dict[int, dict]] # detection 미사용 시 None - is_anomaly_by_cycle_length, is_anomaly_by_vision, is_anomaly_by_omitted_cell, is_anomaly_by_tracked_coordinates: Optional[bool] - anomaly_level: Optional[str] ("ok" / "level-1_anomaly" / "level-2_anomaly") - z_value, sigma, mean, sigma_level, lower_time, upper_time: Optional[float] - centers: Optional[list[tuple[int, float, float] | list[float]]] - average_distance, max_distance, within_score_ratio: Optional[float] - path_template: dict[str, Any] - last_frame: Optional[np.ndarray] - sequence_status: Optional[str] ("ok" / "duplicate" / None) - expected_class_index: Optional[int] — 해당 위치에서 기대한 클래스 - sequence_missing: Optional[list[int]] — 이 사이클까지 건너뛴 미검 클래스 목록

Usage

error, message, cycles = inference_handler.get_cycles()

get_default_inference_option(key)

inference 옵션의 기본 설정 값을 반환합니다.

Parameters:

Name	Type	Description	Default
key	str	옵션 key	required

Returns:

Name	Type	Description
Any	Any	옵션 value

Keys

사용 가능한 key와 value 목록:

{
    "frame_selection_method": "class",  # 프레임 선택 방법
    "frame_selection_queue_size": 16,  # selector queue 길이
    "emit_mode": "current",  # "current" | "previous"
    "lookahead": 0,  # -1(auto) 또는 0 이상
    "use_context_splitter": True,
    # Optional: Energy-based OOD
    "energy_ood_enabled": False,
    "energy_ood_temperature": 1.0,
    "energy_ood_threshold": None,  # float; energy > threshold 이면 OOD
    "energy_ood_fallback_class_index": 0,  # OOD 시 대체 class index
    # Cycle anomaly detection
    "cycle_anomaly_enabled": True,  # on/off
    "cycle_anomaly_mode": "z-test",  # "z-test" | "full_interval" | "marginal_interval"
    "cycle_anomaly_test_mode": "two-sided",  # 전역 검정 방식 "two-sided" | "right-sided" | "left-sided"
    "cycle_anomaly_sigma_level_1": 1.96,  # 전역 95% level-1 임계값
    "cycle_anomaly_sigma_level_2": 2.576,  # 전역 99% level-2 임계값
    # setter-only (set_inference_option으로만 사용):
    # "cycle_anomaly_stats": dict,  # 전체 클래스 통계 {class_id: {mean, sigma, [test_mode, sigma_level_1, sigma_level_2]}, ...}
    # "cycle_anomaly_class_stat": dict,  # 개별 클래스 통계 {class_id, mean, sigma, [test_mode, sigma_level_1, sigma_level_2]}
    # "cycle_anomaly_intervals": dict,  # 전체 interval {class_id: {level_1_lower, ...}, ...}
    # "cycle_anomaly_class_interval": dict,  # 개별 interval {class_id, level_1_lower, ...}
    # "cycle_anomaly_class_test_mode": dict,  # 클래스별 검정 방식 {class_id, test_mode}
    # "cycle_anomaly_class_sigma_level_1": dict,  # 클래스별 level-1 임계값 {class_id, sigma_level_1}
    # "cycle_anomaly_class_sigma_level_2": dict,  # 클래스별 level-2 임계값 {class_id, sigma_level_2}
    # Cycle sequence monitor (실시간 순서 모니터링)
    "sequence_monitor_enabled": True,  # on/off
    "sequence_monitor_expected_sequence": list[int] | None,  # 기대 순서 (read-only)
}

---

get_inference_option(key)

현재 설정된 inference 옵션 값을 읽습니다.

Parameters:

Name	Type	Description	Default
key	str	옵션 key	required

Returns:

Name	Type	Description
Any	Any	옵션 value

Keys

get_default_inference_option과 동일합니다.

---

set_inference_option(key, value)

inference 옵션을 설정합니다.

Parameters:

Name	Type	Description	Default
key	str	옵션 key	required
value	Any	옵션 value	required

Keys

get_default_inference_option과 동일합니다.

---

reset()

InferenceHandler를 초기화합니다. 만약, 영상이 끊겼다가 들어올 경우 사용합니다.

Usage

inference_handler.reset()

---