Monitoring

winding.WindingMonitoringEngine

권취 상태를 감지하기 위한 WindingMonitoringEngine class 입니다.

Usage

expected_output = {
    "state_index": int,
    "is_ng": bool,
    "ng_boxes": (type(None), np.ndarray),
    "require_sponge_wrapping": bool,
    "progress_rate": float,
    "segmentation_contours": {
        "left_drum": (type(None), np.ndarray),
        "right_drum": (type(None), np.ndarray),
        "winding_cable": (type(None), np.ndarray),
        "drum_body": (type(None), np.ndarray)
    },
    "drum_shifting_score": float,
    "cable_shifting_box": (type(None), list),
    "cable_moving_direction": (type(None), str),
    "drum_moving_direction": str,
    "time": (type(None), dict),
}


# 1. WindingMonitoringEngine 빌드
winding_engine_build_config = {
    "segmentation": {
        "checkpoint_path": DEMO_CHECKPOINT,
        "password": PASSWORD,
        "device": DEVICE,
    },
}
error, message, winding_engine = WindingMonitoringEngine.build(winding_engine_build_config)
assert error >= 0, f"[CODE] ({error}) {message}"
print("WindingMonitoringEngine built successfully")

monitoring_options = {
    "params.sponge_wrapping_threshold": 80,
    "params.shift_sensitivity": 50,
    "params.anomaly_sensitivity": 50,
}

# 2. 기본 모니터링 옵션 확인
for key in monitoring_options:
    error, message, value = winding_engine.get_default_monitoring_option(key)
    assert error >= 0, f"[CODE] ({error}) {message}"
    print(f"    Default monitoring option '{key}': {value}")

# 3. 모니터링 옵션 변경
for key, value in monitoring_options.items():
    error, message, _ = winding_engine.set_monitoring_option(key, value)
    assert error >= 0, f"[CODE] ({error}) {message}"

# 4. 모니터링 옵션이 잘 변경되었는지 확인
print("Monitoring options:")
for key in monitoring_options:
    error, message, value = winding_engine.get_monitoring_option(key)
    assert error >= 0, f"[CODE] ({error}) {message}"
    print(f"    {key}: {value}")

# 5. 권취 모니터링 수행
image_paths = os.listdir(DATA)
image_paths = [
    path for path in image_paths if path.lower().endswith((".png", ".jpg", ".jpeg", ".bmp"))
]

for i, image_path in enumerate(image_paths):
    print(f"Processing frame: {image_path}")

    # 프레임 별 검사 중간에 monitoring option 변경.
    if i == len(image_paths) // 2:
        monitoring_options = {
            "params.sponge_wrapping_threshold": 90,
            "params.anomaly_sensitivity": 40,
            "params.shift_sensitivity": 70,
        }
        for key, value in monitoring_options.items():
            error, message, _ = winding_engine.set_monitoring_option(key, value)
            assert error >= 0, f"[CODE] ({error}) {message}"
            print(f"    Monitoring option '{key}' changed to {value}")

    frame = cv2.imread(os.path.join(DATA, image_path))
    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    error, message, result = winding_engine.process_frame(frame)
    assert error >= 0, f"[CODE] ({error}) {message}"

    # 결과 확인
    for key, expected_type in expected_output.items():
        assert key in result, f"Key '{key}' not found in result"
        assert isinstance(result[key], expected_type), f"Key '{key}' has wrong type"
        if isinstance(result[key], dict):
            print(f"    {key}:")
            for _key, _value in result[key].items():
                if isinstance(_value, np.ndarray):
                    print(f"        {_key}: {_value.shape}")
                else:
                    print(f"        {_key}: {_value}")
        elif isinstance(result[key], np.ndarray):
            print(f"    {key}: {result[key].shape}")
        else:
            print(f"    {key}: {result[key]}")

# 6. 새로운 뷰에서 권취 모니터링 수행할 경우,
error, message, _ = winding_engine.reset()
assert error >= 0, f"[CODE] ({error}) {message}"
print("Monitoring in a new view")

for i, image_path in enumerate(image_paths):
    print(f"Processing frame: {image_path}")
    frame = cv2.imread(os.path.join(DATA, image_path))
    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    error, message, results = winding_engine.process_frame(frame)
    assert error >= 0, f"[CODE] ({error}) {message}"

    # 결과 확인
    for key, expected_type in expected_output.items():
        assert key in result, f"Key '{key}' not found in result"
        assert isinstance(result[key], expected_type), f"Key '{key}' has wrong type"
        if isinstance(result[key], dict):
            print(f"    {key}:")
            for _key, _value in result[key].items():
                if isinstance(_value, np.ndarray):
                    print(f"        {_key}: {_value.shape}")
                else:
                    print(f"        {_key}: {_value}")
        elif isinstance(result[key], np.ndarray):
            print(f"    {key}: {result[key].shape}")
        else:
            print(f"    {key}: {result[key]}")

build(config) classmethod

WindingMonitoringEngine class의 instance를 생성합니다.

Parameters:

Name	Type	Description	Default
config	dict	WindingMonitoringEngine를 생성하기 위한 config가 담겨있는 dictionary 입니다. { "segmentation": dict, # Segmentation InferenceHandler를 생성하기 위한 config { "checkpoint_path": str, # segmentation model 체크포인트 경로 "password": str, # 체크포인트 패스워드 "device": Union[int, str], # GPU 번호 (int) or "cpu" (str) (default "cpu") } "monitoring_options": Optional[dict], # 권취 모니터링 옵션 config (default None) # 구조는 `get_default_monitoring_option`의 Keys 참고 }	required

Returns:

Name	Type	Description
WindingMonitoringEngine	WindingMonitoringEngine	build가 완료된 winding.WindingMonitoringEngine class의 instance를 반환합니다.

get_default_monitoring_option(key)

권취 모니터링 옵션의 기본 설정 값을 반환합니다.

Parameters:

Name	Type	Description	Default
key	str	옵션 key	required

Returns:

Name	Type	Description
Any	Any	옵션 value

Keys

사용 가능한 key와 value 목록:

{
    "params.sponge_wrapping_threshold": int, # 스펀지 보양을 시작할 때의 진행률을 결정하는 [1, 100] 사이의 값.
    "params.anomaly_sensitivity": int, # 권취 불량을 감지하기 위한 [1, 100] 사이의 민감도.
    "params.shift_sensitivity": int, # 드럼 이동을 감지하기 위한 [1, 100] 사이의 민감도.
}

사용 예시:

winding_engine.get_default_monitoring_option(key="params.sponge_wrapping_threshold")

get_monitoring_option(key)

현재 설정된 권취 모니터링 옵션 값을 읽습니다.

Parameters:

Name	Type	Description	Default
key	str	옵션 key	required

Returns:

Name	Type	Description
Any	Any	옵션 value

Keys

get_default_monitoring_option의 Keys 참고

set_monitoring_option(key, value)

권취 모니터링 옵션을 설정합니다.

Parameters:

Name	Type	Description	Default
key	str	설정하고자 하는 옵션 key	required
value	Any	성정하고자 하는 옵션 value	required

Returns:

Name	Type	Description
None	None	None

Keys

get_default_monitoring_option의 Keys 참고.

process_frame(frame)

입력 받은 프레임에 대해 권취 모니터링을 수행합니다.

Parameters:

Name	Type	Description	Default
frame	ndarray	권취 모니터링을 수행할 프레임 입니다.	required

Returns:

Name	Type	Description
dict	dict	권취 모니터링 결과를 담고 있는 dictionary 입니다. { "state_index": int, # 입력 받은 프레임의 state index입니다. # 1: 정상적으로 작동 중인 기본 상태입니다. # 2: 드럼 이동이 필요한 상태입니다. # 3: 드럼 사이드 부에 도달하여 방향 전환이 필요한 상태입니다. # 4: 프레임에서 권취 전선이 검출되지 않은 상태입니다. # 5: API 오류가 발생한 상태입니다. "is_ng": bool, # 불량 발생 여부 "ng_boxes": (None, np.ndarray), # 오류가 발생한 영역에 대한 박스들입니다. (N, 4) 형태이며 (x_top, y_left, width, height) 입니다. "require_sponge_wrapping": bool, # 스펀지 보양 작업 필요 여부 "progress_rate": float, # 특정 방향으로의 권취 작업률 입니다. -1.0 또는 [0.0, 1.0] 범위의 실수 값입니다. # 증가할 수록 작업 진행률이 높아 양끝에 가까워 진 상태입니다. "segmentation_contours": { "left_drum": (None, np.ndarray) # 왼쪽 드럼에 대한 contour (N, 2) 입니다. "right_drum": (None, np.ndarray) # 오른쪽 드럼에 대한 contour (N, 2) 입니다. "winding_cable": (None, np.ndarray) # 권취가 완료된 전선에 대한 contour (N, 2) 입니다. "drum_body": (None, np.ndarray) # 권취 전선에 대한 contour (N, 2) 입니다. }, "drum_shifting_score": float, # 드럼을 이동시킬지 여부에 대한 점수입니다. -1.0 또는 [0.0, 1.0] 범위의 실수 값입니다. 증가할 수록 드럼 이동 시점에 가까워진 상태입니다. "cable_shifting_box": (None, list[int]), # 권취 중인 전선이 위치할 수 있는 유효 범위를 나타내는 박스 좌표 [x_top, y_left, width, height] "cable_moving_direction": (None, str), # 권취 중인 전선이 움직이는 방향입니다. ["left_to_right", "right_to_left"] "drum_moving_direction": (None, str), # 드럼이 움직이는 방향입니다. ["left_to_right", "right_to_left"] # cable_moving_direction의 반대 방향에 해당합니다. "time": (None, dict), # 각 단계별 수행 시간입니다. # "segmentation_time": segmentation 수행 시간 (ms) # "direction_monitoring_time": 권취 방향 및 진행률 예측 시간 (ms) # "anomaly_monitoring_time": 권취 불량 예측 시간 (ms) # "shifting_monitoring_time": 드럼 이동 여부 예측 시간 (ms) # "total_time": 전체 수행 시간 (ms) "error_message": (None, str), # API 오류 발생 시 오류 메시지입니다. 정상 작동 시 None입니다. }

reset()

WindingMonitoringEngine을 새롭게 빌드하지 않고, 새로운 뷰에서 모니터링을 할 경우에 process_frame에 앞서 호출해야 합니다.

Returns:

Name	Type	Description
None	None	None