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 입니다. name="__codelineno-0-1" href="#__codelineno-0-1">{ "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}, ...} # 미지정 시 get_inference_option("cycle_anomaly_intervals") / interval 모드에서 stats + sigma_level 기반 파생 구간을 사용 } ``` "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) class="p">}	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`	`int` or `float`	사이클 시작 타임스탬프
`end_frame_ts`	`int` or `float`	사이클 종료 타임스탬프
`class_index`	`int`	클래스 인덱스
`class_type`	`str`	`"normal"` / `"standby"`
`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`	`Optional[bool]`	사이클 길이 이상 여부
`is_anomaly_by_vision`	`Optional[bool]`	비전 이상 여부
`is_anomaly_by_omitted_cell`	`Optional[bool]`	누락 셀 이상 여부
`is_anomaly_by_tracked_coordinates`	`Optional[bool]`	좌표 추적 이상 여부
`anomaly_level`	`Optional[str]`	`"ok"` / `"level-1_anomaly"` / `"level-2_anomaly"`
`z_value`	`Optional[float]`	z-test 값
`sigma`	`Optional[float]`	표준편차
`mean`	`Optional[float]`	평균
`centers`	`Optional[list]`	`tuple[int, float, float]` or `list[float]`
`average_distance`	`Optional[float]`	평균 거리
`max_distance`	`Optional[float]`	최대 거리
`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 임계값
    # get/set 가능 (클래스별 통계/interval 데이터):
    "cycle_anomaly_stats": dict | None,  # 전체 클래스 통계 {class_id: {mean, sigma, [test_mode, sigma_level_1, sigma_level_2]}, ...}
    "cycle_anomaly_class_stat": dict | None,  # set: 개별 클래스 통계 {class_id, mean, sigma, ...} / get: 전체 stats
    "cycle_anomaly_intervals": dict | None,  # 전체 interval {class_id: {level_1_lower, ...}, ...}; 없으면 stats + sigma_level 기반 파생값 반환
    "cycle_anomaly_class_interval": dict | None,  # set: 개별 interval {class_id, level_1_lower, ...} / get: 전체 intervals
    "cycle_anomaly_class_test_mode": dict | None,  # set: {class_id, test_mode} / get: 전체 stats
    "cycle_anomaly_class_sigma_level_1": dict | None,  # set: {class_id, sigma_level_1} / get: 전체 stats
    "cycle_anomaly_class_sigma_level_2": dict | None,  # set: {class_id, sigma_level_2} / get: 전체 stats
    # Cycle sequence monitor (실시간 순서 모니터링)
    "sequence_monitor_enabled": True,  # on/off
    "sequence_monitor_expected_sequence": list[int] | None,  # 기대 순서 (전체, read-only)
    "sequence_monitor_normal_class_sequence": list[int] | None,  # 기대 순서 (standby 제외, 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()

Inference

cycle_counter.CycleCounterInferenceHandler

build(config) classmethod

infer(data, frame_timestamps)

flush_pending_inference()

get_cycles()

get_default_inference_option(key)

get_inference_option(key)

set_inference_option(key, value)

reset()

`cycle_counter.CycleCounterInferenceHandler`

`build(config)` `classmethod`

`infer(data, frame_timestamps)`

`flush_pending_inference()`

`get_cycles()`

`get_default_inference_option(key)`

`get_inference_option(key)`

`set_inference_option(key, value)`

`reset()`