콘텐츠로 이동

Inference

CycleInfo 필드 요약

get_cycles()가 반환하는 cycles는 list[CycleInfo]입니다. 아래 필드 중 일부는 활성화된 모듈이나 후처리 단계에 따라 None이거나 비어 있을 수 있습니다.

기본 메타데이터

키 타입 설명
cycle_length int 사이클 길이
start_frame_ts, end_frame_ts int \| float 사이클 시작/종료 timestamp
class_index int 현재 사이클의 클래스 index
original_class_index Optional[int] 후처리 전 원본 클래스 index
current_model str 사이클을 만든 모델 이름 — build config의 model_name(사용자가 임의로 지정하는 모델/프로젝트 이름). 분류기·detection 사이클 모두 동일
selected_frames list[dict] 프레임 선택기나 detector가 뽑은 대표 프레임 정보
last_frame Optional[np.ndarray \| RawImages] 사이클 마지막 프레임 또는 프레임 묶음

Detection / Path / Vision

키 타입 설명
detections Optional[dict[int, dict]] detection 미사용 시 None, 사용 시 timestamp별 detection 결과
centers Optional[list[tuple[int, float, float] \| list[float]]] 추적 center point 목록
average_distance Optional[float] 기준 path와의 평균 거리
max_distance Optional[float] 기준 path와의 최대 거리
within_score_ratio Optional[float] 허용 영역 내부 비율
path_template dict[str, Any] path 시각화/평가용 template 정보
is_anomaly_by_vision Optional[bool] visual anomaly 후처리 결과
is_anomaly_by_tracked_coordinates Optional[bool] tracked coordinates/path 기반 이상 여부

이상치 / 통계

키 타입 설명
is_anomaly_by_cycle_length Optional[bool] cycle length 기반 이상 여부
is_anomaly_by_omitted_cell Optional[bool] cell checker 기반 이상 여부
cell_check_result Optional[CellCheckResult] 누락 셀 상세 결과
z_value Optional[float] z-test 결과값
sigma Optional[float] 표준편차
mean Optional[float] 평균 cycle length
sigma_level Optional[float] 적용된 sigma level
lower_time, upper_time Optional[float] 허용 cycle time 범위

자주 보는 예시

error_code, error_message, cycles = inference_handler.get_cycles()
first_cycle = cycles[0]

print(first_cycle["cycle_length"])
print(first_cycle["class_index"])
print(first_cycle["is_anomaly_by_cycle_length"])

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],  # 추론 옵션
    "device": Union[int, str],            # GPU 번호 (int) 또는 "cpu" (기본값 "cpu")
    "anomaly_by_cycle_detector": Optional[dict],  # 주기 길이 이상치 검출 설정
    # ── anomaly_by_cycle_detector 내부 ──
    # {
    #     "mode": str,             # "z-test" | "full_interval" | "marginal_interval"
    #     "test_mode": str,        # "two-sided" | "right-sided" | "left-sided"
    #     "sigma_level_1": float,  # level-1 임계 z-값 (95%). 기본값: 1.96
    #     "sigma_level_2": float,  # level-2 임계 z-값 (99%). 기본값: 2.576
    #     "stats": Optional[dict], # {class_id: {mean, sigma, test_mode?, sigma_level_1?, sigma_level_2?}, ...}
    #     "intervals": Optional[dict],  # {class_id: {level_1_lower, level_1_upper, level_2_lower, level_2_upper}, ...}
    # }
    "frame_selector": Optional[dict],     # 프레임 선택 모델 설정
    # ── frame_selector 내부 ──
    # {
    #     "cycle_select_offset_by_class": {class_id: [offset, ...], ...}
    # }
    "detection": Optional[dict],          # detection 엔진 설정
    # ── detection 내부 ──
    # {
    #     "checkpoint_path": str,
    #     "password": Optional[str],
    #     "device": Union[int, str],
    #     "inference_options": Optional[dict],
    # }
    "det_template": Optional[dict | str], # detection tracking 템플릿 (dict 또는 파일 경로)
    # ── det_template 내부 (dict인 경우) ──
    # {
    #     "waypoints": dict[str, list[tuple[int, int]]],
    #     "phase_group_to_class": dict[str, int],
    #     "class_to_phases": dict[int, list[str]],
    # }
    "tracked_anomaly_threshold": float,   # 경로 편차 이상치 판단 임계값 (기본값 5000)
    "use_ts": bool,                       # 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}],
}

Note

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로 대체됩니다.
error int 에러 코드 (0이면 정상)
logit Optional[float] 예측된 class_index에 해당하는 softmax 이전의 원본 logit 값

실제 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 사이클을 만든 모델 이름 — build config의 model_name(사용자가 임의로 지정하는 모델/프로젝트 이름). 분류기·detection 사이클 모두 동일
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과 동일합니다.

get_metadata(class_names=None)

빌드 직후의 클래스별 초기(불변) 메타데이터를 반환합니다.

get_default_inference_option/get_inference_option을 통해 checkpoint에서 읽어들인 sequence/anomaly 설정만으로 클래스별 메타데이터를 구성합니다. set_inference_option으로 stats/intervals/sigma_level 등을 변경하기 전에 호출해야 "최초" 메타데이터를 얻을 수 있습니다.

Parameters:

Name Type Description Default
class_names Optional[dict[int, str]]

class_id → className 매핑. 명시적으로 전달된 항목이 checkpoint metadata에 저장된 매핑보다 우선합니다. 둘 다 없으면 f"class_{class_id}" 형태로 채워집니다.

checkpoint에 매핑을 저장하려면 학습 단계에서 다음과 같이 합니다: Trainer.save_checkpoint(path, metadata={"class_names": {1: "..."}}) (list 형태도 지원되며, index가 class_id로 해석됩니다.)

 None

Returns:

Type Description
list[dict[str, Any]]

list[dict[str, Any]]: 각 클래스에 대해 다음 키를 가지는 dict 목록.
list[dict[str, Any]]

| 키 | 타입 | 설명 |
list[dict[str, Any]]

|---|---|---|
list[dict[str, Any]]

| classId | int | 클래스 인덱스 |
list[dict[str, Any]]

| className | str | 클래스명 (class_names 미전달 시 fallback) |
list[dict[str, Any]]

| classSeq | int | expected_sequence 내의 1-based 위치 |
list[dict[str, Any]]

| isCycleStart | bool | classId가 1이면 True (현재 정책: 항상 classId=1을 cycle start로 간주) |
list[dict[str, Any]]

| sigmaLevel1 | float | level-1 임계값 (per-class 우선, 없으면 전역) |
list[dict[str, Any]]

| sigmaLevel2 | float | level-2 임계값 (per-class 우선, 없으면 전역) |
list[dict[str, Any]]

| lowerLevel1 | Optional[float] | level-1 허용 구간 하한 |
list[dict[str, Any]]

| upperLevel1 | Optional[float] | level-1 허용 구간 상한 |
list[dict[str, Any]]

| lowerLevel2 | Optional[float] | level-2 허용 구간 하한 |
list[dict[str, Any]]

| upperLevel2 | Optional[float] | level-2 허용 구간 상한 |
list[dict[str, Any]]

| baseTime | Optional[float] | 기준 평균 (stats.mean, 없으면 intervals.base_time) |
list[dict[str, Any]]

| isStandby | bool | standby로 분류된 클래스인지 여부 |
Note

checkpoint에 sequence/anomaly 설정이 없는 경우 빈 리스트를 반환합니다.

Usage 
error, message, metadata = inference_handler.get_metadata(
    class_names={1: "center_to_cell_1", 2: "cell_1_to_center", ...}
)

reset()

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

Usage 
inference_handler.reset()