cw2007/move_detect
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
move_detect/tests/test_engine.py
유창욱 4bfb7e0274
Some checks are pending
verify / edge (push) Waiting to run
Details
verify / server (push) Waiting to run
Details
feat: 추락 · 동시 다발 쓰러짐 · 구역 프리셋 (1.2.0) 
AI CCTV 목적을 원론부터 재점검한 결과. 셋 다 새 추론 모델 없이 지금 있는
자세추정·구역·트랙 이력만으로 구현했다.

■ 추락 (fall_from_height)

잡고 있던 것은 같은 높이의 '넘어짐'이었고, 사망 1위 유형인 '떨어짐'은 못
잡았다. GhostIncidentDetector 가 트랙이 이미 위험 자세일 때만 무장하므로
(_DANGEROUS), 발판에 서 있다가 떨어지는 사람은 소실 순간 STANDING 이라
아무것도 안 걸렸다.

판정의 핵심은 bbox 중심이 아니라 하단(y2)이다. 쭈그려 앉으면 중심은
내려가도 발은 그 자리이고, 추락은 발 밑선이 통째로 내려간다 — 중심으로
재면 쭈그려 앉아 작업하는 사람이 전부 추락이 된다. 원근 접근은 y2 도
내려가므로 높이 증가 가드로 따로 막는다. Track.frames 가 90프레임을 들고
있어 별도 버퍼가 없다.

발보는 두 갈래로 나뉘되 한 사고는 한 건이다: 프레임 안에서 넘어짐이
확정되면 유형을 승격하고, 화각 밖으로 사라지면 ghost 무장을 넓혀 그쪽으로
낸다. 심각도 CRITICAL — 척추·두부 손상 주의가 통보에 실리려면 넘어짐(HIGH)과
등급이 갈라져야 한다. 고소 구역은 rules_json 의 elevated 키로 붙어 스키마
변경이 없고, 구역이 없어도 하강 신호만으로 동작한다.

■ 동시 다발 쓰러짐 (multi_down)

같은 구역에서 짧은 시간에 둘이 쓰러진 것은 넘어짐 두 건이 아니라 질식·
가스·폭발이다. 이 기능의 본체는 감지가 아니라 통보 문구다 — 밀폐공간
사망자의 상당수가 구조하러 들어간 사람이므로, 링크를 열기 전에 메일 본문에서
"공기호흡기 없이 진입하지 마십시오"가 읽혀야 한다. Channels::guidance 로
유형별 안내를 붙였고 문구를 테스트로 고정했다(문구가 사라지면 기능이 사라진
것과 같다).

서버에 둔 이유: 가스는 카메라 화각을 존중하지 않아 노드를 넘는 판단이
필요하고, 현장 즉시 대응은 첫 쓰러짐의 경광등이 이미 확보한다. 오탐 방어의
핵심은 근접성 — 넓은 공장에서 2분 안의 무관한 넘어짐 두 건은 실제로
일어나므로 같은 구역이거나 좌표가 가까워야 묶는다. 한 사람이 두 번 넘어진
것은 한 명으로 센다.

■ 구역 프리셋 6종 + max_workers:0 수정

밀폐공간 감시에 필요한 것(제한구역+2인1조+체류상한)은 이미 전부 있었는데
화면이 "최소 인원을 넣으세요"라고만 물어 아무도 그렇게 쓰지 못했다.
안전관리자는 min_workers 가 아니라 밀폐공간을 생각한다.

그 과정에서 잠복 결함 하나: max_workers:0 이 falsy 로 걸러져 "아무도 있으면
안 되는 구역"(크레인 짐 아래·굴착기 선회 반경 = 깔림 사고의 전형)이 통째로
꺼져 있었다. 운영자가 저장해 놓고도 아무 일이 안 일어났다. 감지기·입력란
(min=1)·저장 직렬화(>0)·목록 표시 네 곳에서 막혀 있었다.

■ 검증

pytest 322 passed / phpunit 263 passed. 신규 테스트 30건.
이벤트 유형 2종 추가는 test_event_type_parity 가 강제하는 5곳(엣지 enum ·
EventValidator · schema CHECK · 프런트 라벨 5파일 · 필터 옵션) 전부 동기화했고
기존 설치용 마이그레이션을 추가했다.

■ Pi 번들

1.1.0 과 달리 이번에는 edge/ 가 바뀌었으므로 Pi 번들 재빌드가 필수다.
낡은 1.0.0 번들은 dist/ 에서 제거했다 — 두면 추락 감지가 빠진 노드가 그대로
납품된다. arm64 + 네트워크 빌드 머신에서 재빌드해야 Pi 납품이 가능하다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EBjuR7DW6PNNMows4KnbYU
2026-09-05 12:21:50 +09:00

413 lines
19 KiB
Python
Raw Blame History

from collections import deque
from edge.models import Keypoint, BBox
from edge.tracking import Track, TrackFrame
from edge.events import EventType, Severity
from edge.detectors.fall import FallDetector, FallConfig
from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig
from edge.detectors.ghost import GhostIncidentDetector, GhostConfig
from edge.detectors.zone import ZoneIntrusionDetector, ZoneConfig, Zone
from edge.engine import EventEngine, EngineConfig
def _lying_kps():
kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
kps[0] = Keypoint(90, 320, 0.9)
return kps
def _lying_track(track, t):
track.frames.append(TrackFrame(t, BBox(90, 290, 210, 320), _lying_kps()))
return track
def _standing_kps(cx=150):
kps = [Keypoint(cx, 150, 0.9) for _ in range(17)]
kps[5] = Keypoint(cx - 10, 100, 0.9); kps[6] = Keypoint(cx + 10, 100, 0.9) # shoulders above
kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9) # hips below
kps[0] = Keypoint(cx, 60, 0.9) # head above hips
return kps
def _standing_track(track, t, cx=150):
track.frames.append(TrackFrame(t, BBox(cx - 20, 60, cx + 20, 210), _standing_kps(cx)))
return track
def _engine():
return EventEngine(
FallDetector(FallConfig(prone_confirm_s=0.5)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
ZoneIntrusionDetector([], ZoneConfig()),
EngineConfig(camera_id="cam-01", cooldown_s=30.0),
)
def test_fall_produces_fall_event():
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0)
eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6)
events = eng.process([tr], [], now=0.6)
types = [e.event_type for e in events]
assert EventType.FALL in types
assert events[0].camera_id == "cam-01"
def test_cooldown_suppresses_duplicate():
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # FALL 1회
_lying_track(tr, 0.7)
again = eng.process([tr], [], now=0.7)
assert all(e.event_type != EventType.FALL for e in again)
def test_no_motion_event_carries_duration():
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
ZoneIntrusionDetector([], ZoneConfig()),
EngineConfig(camera_id="cam-01", cooldown_s=30.0),
)
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # -> fall
_lying_track(tr, 1.0); eng.process([tr], [], now=1.0) # still
_lying_track(tr, 2.0)
events = eng.process([tr], [], now=2.0) # immobile >1.0s -> no_motion
no_motion_events = [e for e in events if e.event_type == EventType.NO_MOTION]
assert len(no_motion_events) == 1
e = no_motion_events[0]
assert e.duration_s > 0
assert e.duration_s == e.metrics["immobility_s"]
def test_forget_prunes_state_on_lost():
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6)
events = eng.process([tr], [], now=0.6) # FALL fires
assert any(e.event_type == EventType.FALL for e in events)
assert (EventType.FALL, 1) in eng._last_emit
assert 1 in eng.fall._states
# track is now lost -> engine must prune its per-track state
eng.process([], [tr], now=1.0)
assert not any(len(k) >= 2 and k[1] == 1 for k in eng._last_emit)
assert 1 not in eng.fall._states
def test_pinch_zone_emits_pinch_event_not_warning():
# A 협착(pinch) zone is the highest-danger zone type (machinery entrapment). It must fire
# as PINCH/HIGH, not be silently downgraded to a generic ZONE_WARNING.
poly = [(100, 100), (200, 100), (200, 200), (100, 200)]
zdet = ZoneIntrusionDetector([Zone("z1", "pinch", poly)], ZoneConfig(debounce_s=0.0))
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=999)), # keep fall/no_motion quiet
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
zdet, EngineConfig(camera_id="cam-01", cooldown_s=30.0))
kps = [Keypoint(150, 150, 0.9) for _ in range(17)] # both ankles inside the zone
tr = Track(2, deque(maxlen=90))
tr.frames.append(TrackFrame(0.0, BBox(140, 140, 160, 160), kps))
events = eng.process([tr], [], now=0.0)
pinch = [e for e in events if e.event_type == EventType.PINCH]
assert len(pinch) == 1
assert pinch[0].severity == Severity.HIGH
assert pinch[0].zone_id == "z1"
assert not any(e.event_type == EventType.ZONE_WARNING for e in events)
def test_ghost_emit_does_not_leak_a_permanent_dedup_key():
# A GHOST fires reappear_s AFTER the track was lost (and already forgotten). Routing it through
# the engine cooldown used to insert a (GHOST, track_id) key that forget() could never reclaim
# (the track never appears in lost_tracks again) — a slow 24/7 memory leak. Emitting directly
# must leave no such key behind.
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # confirmed fallen -> dangerous state
eng.process([], [tr], now=1.0) # lost -> ghost candidate registered
events = eng.process([], [], now=12.0) # > reappear_s (10s), no reappearance
assert any(e.event_type == EventType.GHOST for e in events)
assert not any(k[0] == EventType.GHOST for k in eng._last_emit) # no leaked ghost dedup key
def test_two_simultaneous_tracks_have_independent_fall_state():
# Per-track detector state must not leak between two people monitored at once: a fall on track A
# must arm and emit for A only, never for a standing track B. All other engine/detector tests
# exercise a single person, so this pins the multi-person isolation the fall/immobility/zone
# timers rely on (they key every bit of state off track_id).
eng = _engine() # prone_confirm_s=0.5
a = Track(1, deque(maxlen=90)); b = Track(2, deque(maxlen=90))
_lying_track(a, 0.0); _standing_track(b, 0.0, cx=500)
eng.process([a, b], [], now=0.0)
_lying_track(a, 0.6); _standing_track(b, 0.6, cx=500)
events = eng.process([a, b], [], now=0.6)
falls = [e for e in events if e.event_type == EventType.FALL]
assert len(falls) == 1 # exactly one fall...
assert falls[0].track_id == 1 # ...for the lying track A, not the standing track B
def test_fall_fires_at_exactly_prone_confirm_s_inclusive_boundary():
# The confirm is `now - prone_since >= prone_confirm_s`, so a fall must fire at EXACTLY the
# threshold, not one frame later. Pins the inclusive boundary against a >= -> > regression that
# would silently delay every fall confirmation by a frame — up to +0.5s at a thermally throttled
# 2 fps, on the single most safety-critical event. Existing tests only feed comfortably-past
# values (0.6 vs 0.5), so none of them would catch that shift.
eng = _engine() # prone_confirm_s=0.5
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0) # prone_since = 0.0
_lying_track(tr, 0.5) # exactly prone_confirm_s elapsed
events = eng.process([tr], [], now=0.5)
assert any(e.event_type == EventType.FALL for e in events)
def test_occurred_at_uses_wall_clock_while_timers_use_monotonic():
# A fall is CONFIRMED by the monotonic timer (prone_confirm_s over the injected `now`), but the
# event must be STAMPED with wall time so an NTP step does not mis-timestamp evidence (and the
# server's wall-time retention purge does not delete it). Timer `now` and `wall_now` are
# independent axes here to prove occurred_at follows wall time, not the monotonic clock.
eng = _engine() # prone_confirm_s=0.5
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0)
eng.process([tr], [], now=0.0, wall_now=1_700_000_000.0)
_lying_track(tr, 0.6)
events = eng.process([tr], [], now=0.6, wall_now=1_700_000_000.6)
fall = [e for e in events if e.event_type == EventType.FALL]
assert len(fall) == 1
assert fall[0].occurred_at == 1_700_000_000.6 # wall time, NOT the monotonic 0.6
def test_process_defaults_occurred_at_to_now_when_no_wall_clock():
# Single-clock callers (unit tests, direct callers) omit wall_now — occurred_at must fall back
# to the timer `now` so existing behavior is unchanged.
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6)
events = eng.process([tr], [], now=0.6)
fall = [e for e in events if e.event_type == EventType.FALL]
assert len(fall) == 1 and fall[0].occurred_at == 0.6
def test_lone_worker_rule_fires_through_engine():
# End-to-end wiring of the occupancy rules: a 2인 1조 zone with one standing worker inside
# past the confirm window must emit LONE_WORKER/HIGH with the zone id attached.
from edge.detectors.occupancy import OccupancyWatchdog, OccupancyConfig
poly = [(400, 100), (600, 100), (600, 300), (400, 300)]
zdet = ZoneIntrusionDetector([Zone("z1", "restricted", poly, {"min_workers": 2})],
ZoneConfig(debounce_s=0.0))
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=999)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
zdet, EngineConfig(camera_id="cam-01", cooldown_s=30.0),
occupancy=OccupancyWatchdog(OccupancyConfig(lone_confirm_s=5.0)))
tr = Track(1, deque(maxlen=90))
_standing_track(tr, 0.0, cx=500) # ankles at (490,200)/(510,200) — inside the zone
events = eng.process([tr], [], now=0.0)
assert not any(e.event_type == EventType.LONE_WORKER for e in events)
_standing_track(tr, 6.0, cx=500)
events = eng.process([tr], [], now=6.0)
lone = [e for e in events if e.event_type == EventType.LONE_WORKER]
assert len(lone) == 1
assert lone[0].severity == Severity.HIGH
assert lone[0].zone_id == "z1" and lone[0].track_id == 1
assert lone[0].metrics["required"] == 2
def test_fall_event_carries_normalized_foot_spatial():
# Every per-track event now carries the foot point in the zones' 0..1 space — the raw
# material of the server-side heatmap. frame_wh drives the normalization.
eng = _engine() # prone_confirm_s=0.5
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0, frame_wh=(640, 360))
_lying_track(tr, 0.6)
events = eng.process([tr], [], now=0.6, frame_wh=(640, 360))
fall = [e for e in events if e.event_type == EventType.FALL][0]
assert fall.spatial is not None
x, y = fall.spatial
assert 0.0 <= x <= 1.0 and 0.0 <= y <= 1.0
def test_near_fall_emits_info_event():
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=5.0, falling_confirm_s=0.5, recover_grace_s=1.0)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
ZoneIntrusionDetector([], ZoneConfig()),
EngineConfig(camera_id="cam-01", cooldown_s=30.0))
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
_lying_track(tr, 1.0); eng.process([tr], [], now=1.0) # FALLING
_standing_track(tr, 2.0); eng.process([tr], [], now=2.0) # 회복 시작
_standing_track(tr, 3.5)
events = eng.process([tr], [], now=3.5) # grace 경과 → near_fall
near = [e for e in events if e.event_type == EventType.NEAR_FALL]
assert len(near) == 1
assert near[0].severity == Severity.INFO
assert near[0].metrics["prone_s"] == 2.0
def test_ghost_spatial_is_normalized_with_frame_size():
eng = _engine()
tr = Track(1, deque(maxlen=90))
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0, frame_wh=(640, 360))
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6, frame_wh=(640, 360)) # FALLEN
eng.process([], [tr], now=1.0, frame_wh=(640, 360)) # lost while dangerous
events = eng.process([], [], now=12.0, frame_wh=(640, 360))
ghost = [e for e in events if e.event_type == EventType.GHOST][0]
x, y = ghost.spatial
assert 0.0 <= x <= 1.0 and 0.0 <= y <= 1.0 # not raw pixels (150, 305)
def test_ergo_report_emitted_anonymous_and_info():
from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=999)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
ZoneIntrusionDetector([], ZoneConfig()),
EngineConfig(camera_id="cam-01"),
ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.5)))
tr = Track(1, deque(maxlen=90))
_standing_track(tr, 0.0); eng.process([tr], [], now=0.0)
_standing_track(tr, 0.6)
events = eng.process([tr], [], now=0.6)
rep = [e for e in events if e.event_type == EventType.ERGO_REPORT]
assert len(rep) == 1
assert rep[0].severity == Severity.INFO
assert rep[0].track_id is None and rep[0].spatial is None # 익명 합계: 개인 좌표 없음
assert rep[0].metrics["observed_s"] == 0.6
def test_incident_state_track_does_not_accrue_ergo_exposure():
# 쓰러진 사람의 90° 몸통은 인체공학 노출이 아니라 사고다 — FALLING 이후는 집계 제외.
from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
eng = EventEngine(
FallDetector(FallConfig(prone_confirm_s=999, falling_confirm_s=0.05)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig()),
ZoneIntrusionDetector([], ZoneConfig()),
EngineConfig(camera_id="cam-01"),
ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.2)))
tr = Track(1, deque(maxlen=90))
for t in (0.0, 0.1, 0.2, 0.3):
_lying_track(tr, t)
events = eng.process([tr], [], now=t)
assert not any(e.event_type == EventType.ERGO_REPORT for e in events)
def test_allow_cooldown_primitive():
eng = _engine() # EngineConfig(camera_id="cam-01", cooldown_s=30.0)
key = ("FALL", 1)
assert eng._allow(key, now=0.0) is True # first time allowed
assert eng._allow(key, now=5.0) is False # within cooldown -> suppressed
assert eng._allow(key, now=40.0) is True # after cooldown_s elapsed -> allowed again
# a different key is independent
assert eng._allow(("FALL", 2), now=5.0) is True
# --- 추락 (fall_from_height) ----------------------------------------------------------
#
# 같은 높이의 '넘어짐'과 높이에서의 '떨어짐'은 다른 사건이다. 떨어짐은 척추 손상 가능성이
# 있어 함부로 옮기면 안 되고, 그 주의가 통보에 실리려면 유형과 심각도가 갈라져야 한다.
# 발보 경로는 두 갈래다: 프레임 안에서 확정 / 화각 밖으로 사라짐.
from edge.detectors.height import DescentWatch, DescentConfig
def _height_engine(zones=None):
return EventEngine(
FallDetector(FallConfig(prone_confirm_s=0.5)),
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
GhostIncidentDetector(GhostConfig(reappear_s=1.0)),
ZoneIntrusionDetector(zones or [], ZoneConfig()),
EngineConfig(camera_id="cam-01", cooldown_s=30.0),
descent=DescentWatch(DescentConfig(drop_ratio=1.2, drop_window_s=1.5, link_s=3.0)),
)
def _falling_track(track, t, y2, height=150.0, cx=150.0):
"""발 밑선 y2 로 서 있는 사람 프레임을 붙인다."""
track.frames.append(TrackFrame(t, BBox(cx - 20, y2 - height, cx + 20, y2),
_standing_kps(int(cx))))
return track
def test_visible_drop_then_prone_is_fall_from_height_not_fall():
eng = _height_engine()
tr = Track(1, deque(maxlen=90))
_falling_track(tr, 0.0, y2=200.0) # 발판 위
_falling_track(tr, 0.5, y2=320.0)
_falling_track(tr, 1.0, y2=420.0) # 220px 하강 = 키의 1.47배
eng.process([tr], [], now=1.0)
_lying_track(tr, 1.2)
eng.process([tr], [], now=1.2) # prone 시작
_lying_track(tr, 2.0)
out = eng.process([tr], [], now=2.0) # 0.5초 지속 -> 확정
types = [e.event_type for e in out]
assert EventType.FALL_FROM_HEIGHT in types, types
assert EventType.FALL not in types, "한 번의 사고가 두 건으로 올라가면 안 된다"
ev = next(e for e in out if e.event_type == EventType.FALL_FROM_HEIGHT)
assert ev.severity == Severity.CRITICAL
assert ev.metrics["drop_ratio"] >= 1.2
def test_plain_fall_without_descent_stays_fall():
eng = _height_engine()
tr = Track(2, deque(maxlen=90))
_lying_track(tr, 0.0)
eng.process([tr], [], now=0.0)
_lying_track(tr, 0.6)
out = eng.process([tr], [], now=0.6)
assert [e.event_type for e in out] == [EventType.FALL]
def test_drop_then_vanish_emits_fall_from_height_not_ghost():
# 화각 밖으로 떨어진 사람에게는 이쪽이 유일한 발보 경로다.
eng = _height_engine()
tr = Track(3, deque(maxlen=90))
_falling_track(tr, 0.0, y2=200.0)
_falling_track(tr, 0.5, y2=330.0)
_falling_track(tr, 0.9, y2=430.0)
eng.process([tr], [], now=0.9) # 하강 확정 = ghost 무장
eng.process([], [tr], now=1.0) # 소실
out = eng.process([], [], now=3.0) # 재출현 대기 초과
types = [e.event_type for e in out]
assert EventType.FALL_FROM_HEIGHT in types, types
assert EventType.GHOST not in types
ev = next(e for e in out if e.event_type == EventType.FALL_FROM_HEIGHT)
assert ev.metrics["exited_frame"] is True
assert ev.metrics["armed_by"] == "descent"
def test_standing_person_vanishing_without_descent_is_not_a_height_fall():
# 장애물 뒤로 걸어 들어간 사람까지 추락으로 올리면 안 된다.
eng = _height_engine()
tr = Track(4, deque(maxlen=90))
_falling_track(tr, 0.0, y2=200.0)
_falling_track(tr, 1.0, y2=205.0)
eng.process([tr], [], now=1.0)
eng.process([], [tr], now=1.1)
out = eng.process([], [], now=3.0)
assert [e.event_type for e in out] == []
def test_elevated_zone_arms_without_a_visible_drop():
# 발판 위에서 화각 밖으로 사라지면 하강을 못 봤어도 추락 후보다.
poly = [(0.0, 0.0), (400.0, 0.0), (400.0, 400.0), (0.0, 400.0)]
zones = [Zone("zhigh", "restricted", poly, {"elevated": True})]
eng = _height_engine(zones)
tr = Track(5, deque(maxlen=90))
_falling_track(tr, 0.0, y2=200.0)
_falling_track(tr, 1.0, y2=205.0)
eng.process([tr], [], now=1.0)
eng.process([], [tr], now=1.1)
out = eng.process([], [], now=3.0)
fh = [e for e in out if e.event_type == EventType.FALL_FROM_HEIGHT]
assert len(fh) == 1, [e.event_type for e in out]
assert fh[0].metrics["armed_by"] == "zone"
Reference in a new issue View git blame Copy permalink