콘텐츠로 이동

InferenceHandler

safety.engine.api.InferenceHandler

NPU edge 추론 기반으로 Safety 기능을 수행하는 InferenceHandler class 입니다.

Usage

demo/inference.py 참고

build(config) classmethod

InferenceHandler instance를 생성합니다.

Parameters:

Name Type Description Default
config dict

InferenceHandler를 build 하기 위한 config dictionary.

{
    # inference_handler config
    "detection_checkpoint_path": str,         # Detection 모델 (.saigeedge) 경로
    "classification_checkpoint_path": str,    # Classification 모델 (.saigeedge) 경로
    "reid_checkpoint_path": str,              # ReID 모델 (.saigeedge) 경로
    "password": str | None,                   # AES password.

    # event_alarm_handlers config
    "camera_ids": list[int] | None, # 검사할 camera_id list (default None).
}

required

Returns:

Name Type Description
InferenceHandler InferenceHandler

build가 완료된 instance.

Note

default_alarm_options, default_object_options는 API에서 기본적으로 비활성화합니다 (enabled=False).

Usage
err, msg, handler = InferenceHandler.build({
    "detection_checkpoint_path": "/path/to/det.saigeedge",
    "classification_checkpoint_path": "/path/to/cls.saigeedge",
    "reid_checkpoint_path": "/path/to/reid.saigeedge",
    "password": "secret",
    "camera_ids": [0],
})

infer(images, camera_ids, timestamps)

현재 설정된 inference 옵션을 바탕으로 inference를 수행합니다.

Parameters:

Name Type Description Default
images list[ndarray]

inference를 수행할 BGR uint8 image list (shape: [H, W, 3])

required
camera_ids list[int]

images의 각 image에 대한 camera_id list

required
timestamps list[int]

images의 각 image에 대한 timestamp list (단위: 밀리초)

required

Returns:

Type Description
list[dict]

list[dict]: 검사 결과

[
    {
        "detected_objects": [ # 검출된 object들에 대한 정보 리스트
            {
                "id": int | None, # tracker가 부여한 object ID. tracking 미확정 시 None.
                "bounding_box": list[int], # object bounding box (x, y, w, h)
                "bounding_box_norm": list[float], # bounding_box 를 frame 경계로 clip 후
                #   원본 이미지 크기로 0~1 정규화한 [x, y, w, h]. frame 내부 보장 (x+w<=1, y+h<=1).
                "object_class": str, # object의 class 이름.
                "object_score": int, # object의 det score. (범위: 0 ~ 100)
                "movement_score": int, # tracker가 추정한 움직임 정도. (범위: 0 ~ 100)
                "class_index": int, # 0=person, 1=fire, 2=smoke, 3=excavator, 4=forklift, 5=crane_body, 6=crane_hook, 7=crane_lifted_object, 8=heavy_equipment, 9=vehicle
                "classification_scores": {
                    "helmet": float,        # 헬멧 착용 정도 (범위: 0 ~ 100)
                    "head": float,          # 머리 가시성 (범위: 0 ~ 100)
                    "person": float,        # person confidence (범위: 0 ~ 100)
                    "fire": float,          # 화재 정도 (범위: 0 ~ 100)
                    "smoke": float,         # 연기 정도 (범위: 0 ~ 100)
                    "fall_down": float,     # 쓰러짐 정도 (범위: 0 ~ 100)
                    "harness": float,       # 안전벨트 착용 정도 (범위: 0 ~ 100)
                } | None, # person / fire / smoke object 에 존재. 그 외 None.
                # 아래 세 키는 person object 에 한정.
                "brightness_score": int,   # bbox 영역 grayscale mean (0 ~ 255).
                "surrounding_brightness_score": int,  # bbox 주변 ring grayscale mean (0 ~ 255).
                "is_silhouetted": bool,    # small + dark + dark-surrounding 으로 판정된
                                           # silhouette person. NoHelmet/NoHarness/FallDown
                                           # event 는 silhouette 에서 false-positive 회피
                                           # 위해 발화 안 함. dark check 가 비활성이면 False.
                "events": {
                    "without_helmet": bool,  # 헬멧 미착용 (person object 한정)
                    "without_harness": bool, # 안전벨트 미착용 (person object 한정)
                    "fallen_person": bool,   # 쓰러짐 감지 (person object 한정)
                    "fire": bool,            # 화재 감지 (fire object 한정)
                    "smoke": bool,           # 연기 감지 (smoke object 한정)
                    "trespass": bool, # 침입금지구역 진입 (person object 한정)
                    "in_danger_zone": bool,   # 위험구역 진입 (person object 한정)
                    "crush_hazard": bool,    # 중장비 협착 위험 (person object 한정)
                    "collision_hazard": bool, # 중장비 충돌 위험 (person object 한정)
                },
                "alarms": { # object에 발생한 각 event에 대한 alarm 발화 여부.
                    "without_helmet": bool,
                    "without_harness": bool,
                    "fallen_person": bool,
                    "fire": bool,
                    "smoke": bool,
                    "trespass": bool,
                    "in_danger_zone": bool,
                    "crush_hazard": bool,
                    "collision_hazard": bool,
                },
                "event_status": { # event 중간 상태 문자열. 각 event에 대한 alarm이 비활성화된 경우엔 None.
                    "without_helmet": str | None,
                    "without_harness": str | None,
                    "fallen_person": str | None,
                    "fire": str | None,
                    "smoke": str | None,
                    "trespass": str | None,
                    "in_danger_zone": str | None,
                    "crush_hazard": str | None,
                    "collision_hazard": str | None,
                },
            }, ... # 해당 image에서 검출된 object 수만큼 반복.
        ],
        "time": { # 해당 image의 추론에 소요된 시간을 담고있는 dictionary. (단위: 밀리초)
            "preprocess_time": float,       # 이미지 및 기타 전처리에 소요된 시간
            "inference_det_time": float,    # detection 추론에 소요된 시간
            "postprocess_det_time": float,  # detection 후처리에 소요된 시간
            "inference_cls_time": float,    # classification 추론에 소요된 시간
            "postprocess_cls_time": float,  # classification 후처리에 소요된 시간
            "tracker_time": float,          # tracker에 소요된 시간
            "event_alarm_time": float,      # event/alarm 처리에 소요된 시간
        },
    }, ... # 들어온 image 수 만큼 반복 (들어온 image 순서대로)
]

Raises:

Type Description
InvalidNumberOfInputsError

images, camera_ids, timestamps의 길이가 서로 다를 때

UnregisteredCameraIdError

등록되지 않은 camera_id가 포함된 경우

InvalidTimestampError

timestamp가 음수이거나 같은 camera_id에 대해 중복일 때

Usage
err, msg, results = handler.infer(
    images=[frame_a, frame_b],
    camera_ids=[0, 0],
    timestamps=[0, 100],
)
for result in results:
    for obj in result["detected_objects"]:
        print(obj["object_class"], obj["events"])

warmup()

더미 입력으로 NPU pipeline을 사전 warmup 합니다.

Usage
err, msg, _ = handler.warmup()

register_camera(camera_id)

검사할 camera_id를 등록합니다.

Parameters:

Name Type Description Default
camera_id int

등록할 camera_id

required

Raises:

Type Description
InvalidCameraIdError

camera_id가 비음수 int가 아닐 때

AlreadyRegisteredCameraIdError

이미 등록된 camera_id일 때

Usage
err, msg, _ = handler.register_camera(camera_id=0)
err, msg, cam_ids = handler.get_registered_camera_ids()
0 in cam_ids
# True

release_camera(camera_id)

등록된 camera_id를 해제합니다.

Parameters:

Name Type Description Default
camera_id int

해제할 camera_id

required

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, _ = handler.release_camera(camera_id=0)
err, msg, cam_ids = handler.get_registered_camera_ids()
0 in cam_ids
# False

get_registered_camera_ids()

등록된 camera_id 리스트를 반환합니다.

Returns:

Type Description
list[int]

list[int]: 등록된 camera_id 리스트

Usage
err, msg, cam_ids = handler.get_registered_camera_ids()
print(cam_ids)
# [0]

get_default_tracker_option(camera_id)

기본 tracker option을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

사용하지 않음, API 일관성을 위해 존재.

required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: tracker option

Usage
err, msg, option = handler.get_default_tracker_option(camera_id=0)
print(option)
# {"object_score_threshold": 70, "min_continuous_count": 1}

get_tracker_option(camera_id)

등록된 camera_id에 대한 tracker option을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

tracker option을 반환할 camera_id

required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: tracker option

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, option = handler.get_tracker_option(camera_id=0)
print(option)
# 사용자가 설정한 값이 반환됩니다.

set_tracker_option(camera_id, tracker_option)

등록된 camera_id에 대한 tracker option을 설정합니다.

Parameters:

Name Type Description Default
camera_id int

tracker option을 설정할 camera_id

required
tracker_option dict[str, Any]

tracker option

required

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, option = handler.get_tracker_option(camera_id=0)
option["object_score_threshold"] = 50
option["min_continuous_count"] = 3
err, msg, _ = handler.set_tracker_option(camera_id=0, tracker_option=option)

get_default_object_options(camera_id)

기본 object options을 반환합니다 (enabled=False 강제).

Parameters:

Name Type Description Default
camera_id int

사용하지 않음, API 일관성을 위해 존재.

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: 각 object class별 default option dict

Usage
err, msg, options = handler.get_default_object_options(camera_id=0)
print(options["person"])
# {"enabled": False, "area_threshold": 900.0, "score_threshold": 30}
Note

enabled=False 인 class 의 object 는 event/alarm 평가 단계에서 drop — events / alarms / event_status 키가 추가되지 않습니다. detection 자체는 detected_objects 에 그대로 남아 caller 가 raw bbox 를 render 할 수 있습니다. area_threshold 미만 / score_threshold 미만 / frame 면적의 80% 초과 도 동일하게 drop.

get_object_options(camera_id)

등록된 camera_id에 대한 object options을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

object options을 반환할 camera_id

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: object options

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_object_options(camera_id=0)
print(options["person"]["enabled"])
# False  (build / register_camera 직후엔 모두 False)

set_object_options(camera_id, object_options)

등록된 camera_id에 대한 object options을 설정합니다.

Parameters:

Name Type Description Default
camera_id int

object options을 설정할 camera_id

required
object_options dict[str, dict]

object options

required

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_object_options(camera_id=0)
options["person"]["enabled"] = True
err, msg, _ = handler.set_object_options(camera_id=0, object_options=options)

get_default_event_options(camera_id)

기본 event options을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

사용하지 않음, API 일관성을 위해 존재.

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: API event type을 key로 하는 default event options

Usage
err, msg, options = handler.get_default_event_options(camera_id=0)
print(options["without_helmet"])

get_event_options(camera_id)

등록된 camera_id에 대한 event options을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

event options을 반환할 camera_id

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: API event type을 key로 하는 event options

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_event_options(camera_id=0)
print(options["fire"])

set_event_options(camera_id, event_options)

등록된 camera_id에 대한 event options을 설정합니다.

Parameters:

Name Type Description Default
camera_id int

event options을 설정할 camera_id

required
event_options dict[str, dict]

API event type을 key로 하는 event options

required

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_event_options(camera_id=0)
options["without_helmet"]["helmet_score_threshold"] = 40
err, msg, _ = handler.set_event_options(camera_id=0, event_options=options)

get_default_alarm_options(camera_id)

기본 alarm options을 반환합니다 (enabled=False 강제).

Parameters:

Name Type Description Default
camera_id int

사용하지 않음, API 일관성을 위해 존재.

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: API event type을 key로 하는 default alarm options

Usage
err, msg, options = handler.get_default_alarm_options(camera_id=0)
print(options["without_helmet"])
# {"enabled": False, ...}

get_alarm_options(camera_id)

등록된 camera_id에 대한 alarm options을 반환합니다.

Parameters:

Name Type Description Default
camera_id int

alarm options을 반환할 camera_id

required

Returns:

Type Description
dict[str, dict]

dict[str, dict]: API event type을 key로 하는 alarm options

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_alarm_options(camera_id=0)
print(options["fire"]["enabled"])
# False  (build / register_camera 직후엔 모두 False)

set_alarm_options(camera_id, alarm_options)

등록된 camera_id에 대한 alarm options을 설정합니다.

Parameters:

Name Type Description Default
camera_id int

alarm options을 설정할 camera_id

required
alarm_options dict[str, dict]

API event type을 key로 하는 alarm options

required

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id일 때

Usage
err, msg, options = handler.get_alarm_options(camera_id=0)
options["without_helmet"]["enabled"] = True
options["without_helmet"]["alarm_interval"] = 60
err, msg, _ = handler.set_alarm_options(camera_id=0, alarm_options=options)

get_monitoring_area(camera_id)

등록된 camera_id 에 대한 monitoring area 를 반환합니다.

Parameters:

Name Type Description Default
camera_id int

monitoring area 를 반환할 camera_id.

required

Returns:

Type Description
dict[str, list[list[list[int]]]]

dict[str, list[list[list[int]]]]: dictionary of monitoring area polygon list

{
    "includes": [  # 포함하는 영역의 polygon 리스트 (default []).
        [[x1, y1], ..., [xn, yn]],  # n>=3 점으로 이루어진 polygon.
        ...
    ],
    "excludes": list[list[list[int]]],  # 제외하는 영역의 polygon 리스트.
}

Raises:

Type Description
UnregisteredCameraIdError

등록되지 않은 camera_id 일 때.

Usage
err, msg, area = handler.get_monitoring_area(camera_id=0)
print(area)
# {"includes": [], "excludes": []}  # build / register_camera 직후 기본값.

set_monitoring_area(camera_id, monitoring_area)

등록된 camera_id 에 대한 monitoring area 를 설정합니다.

includes 가 비어있으면 (또는 키 자체 미지정) include 검사는 스킵되어 모든 obj 가 통과합니다. excludes 가 비어있으면 exclude 검사는 스킵됩니다. 둘 다 비어있으면 filter 는 no-op 입니다.

Parameters:

Name Type Description Default
camera_id int

monitoring area 를 설정할 camera_id.

required
monitoring_area dict[str, list[list[list[int]]]]

형태는 :meth:get_monitoring_area 반환값과 동일.

required

Raises:

Type Description
InvalidMonitoringAreaError

monitoring_area dict 가 아니거나 polygon 형식이 잘못됐을 때.

UnregisteredCameraIdError

등록되지 않은 camera_id 일 때.

Usage
area = {"includes": [[[0, 0], [100, 0], [100, 100], [0, 100]]], "excludes": []}
err, msg, _ = handler.set_monitoring_area(camera_id=0, monitoring_area=area)

get_default_inference_option(model, key)

(model, key) 에 대한 inference option 의 기본값을 반환합니다.

Inference option 은 모델 재컴파일 없이 런타임에 후처리 knob 을 조정하는 global API 입니다 (camera_id 없음). 지원 범위는 다음과 같습니다.

Parameters:

Name Type Description Default
model str

"safety" 또는 "detection". "classification" 은 Edge 의 MLC sigmoid CLS 에는 softmax 전용 additional_scores 개념이 없어 미지원이며, "depth" / "match" 는 depth 모델 미사용으로 미지원입니다.

required
key str

옵션 키. 가능한 (model, key) 조합:

{
    ("safety", "outputs.time"): bool,
        # 인퍼런스 결과에 stage 별 timing dict 를 포함할지 여부 (default False).
    ("safety", "params.mlc_filter_thresholds"): dict[str, float],
        # MLC 필터 임계값. keys = {"fire", "smoke"}, values [0, 1].
        # (default {"fire": 0.5, "smoke": 0.5}).
    ("detection", "params.object_score_threshold"): list[int],
        # 각 클래스의 score threshold (0-100). DET score 가 이 값보다 작으면 drop.
        # 길이 10 (DetectionClassIndices 순서). default [30] + [10] * 9.
    ("detection", "params.object_area_threshold"): list[int],
        # 각 클래스의 area threshold (>= 0). bbox 면적이 이 값보다 작으면 drop.
        # 길이 10. default [0] * 10.
    ("detection", "params.max_num_of_detected_objects"): list[int],
        # 각 클래스의 최대 detection 개수. 초과 시 score 내림차순 top-N 만 유지.
        # 각 값 >= -1 (-1 = unlimited). 길이 10. default [30] + [-1] * 9.
}

required

Returns:

Name Type Description
Any Any

(model, key) 의 기본값 사본.

Raises:

Type Description
InvalidInferenceOptionError

지원하지 않는 (model, key) 조합.

Usage
err, msg, value = handler.get_default_inference_option(
    model="detection", key="params.max_num_of_detected_objects"
)
print(value)  # [30, -1, -1, -1, -1, -1, -1, -1, -1, -1]

get_inference_option(model, key)

(model, key) 에 대한 현재 설정된 inference option 값을 반환합니다.

Parameters:

Name Type Description Default
model str

:meth:get_default_inference_option 참고.

required
key str

옵션 키. :meth:get_default_inference_option 와 동일.

required

Returns:

Name Type Description
Any Any

현재 값의 사본.

Raises:

Type Description
InvalidInferenceOptionError

지원하지 않는 (model, key) 조합.

Usage
err, msg, value = handler.get_inference_option(
    model="safety", key="outputs.time"
)
print(value)  # False (default)

set_inference_option(model, key, value)

(model, key) 에 대한 inference option 을 설정합니다.

Parameters:

Name Type Description Default
model str

:meth:get_default_inference_option 참고.

required
key str

옵션 키. :meth:get_default_inference_option 와 동일.

required
value Any

설정할 값. 형식/범위는 get_default_inference_option 의 스키마를 따름.

required

Raises:

Type Description
InvalidInferenceOptionError

지원하지 않는 (model, key) 또는 잘못된 값 (타입/범위/길이 위반).

InvalidMlcFilterThresholdError

("safety", "params.mlc_filter_thresholds") 의 dict 가 {person, fire, smoke} 키 또는 [0, 1] 범위를 어길 때.

Usage
err, msg, _ = handler.set_inference_option(
    model="detection", key="params.object_score_threshold",
    value=[50] + [10] * 9,
)

reset()

모든 camera의 tracker / alarm state를 초기화합니다.

Usage
err, msg, _ = handler.reset()

release()

모든 inferencer가 잡고 있는 NPU backend 리소스를 해제합니다.

Usage
err, msg, _ = handler.release()

# 또는 context manager 패턴 권장:
with handler:
    err, msg, results = handler.infer(...)
# __exit__에서 release() 자동 호출