Devin.KR

이동평균과 지연

80분 안팎

학습 목표

고정 입력에서 이동평균 창 크기별 분산 감소와 계단 응답 지연을 비교합니다.

개념

부드러운 신호가 늦은 정지를 만들 수 있습니다

거리 입력이 흔들리면 정지 문턱을 여러 번 넘나들 수 있습니다. 이동평균을 넣으면 값이 안정되어 보이지만 장애물이 갑자기 가까워져도 오래된 먼 거리 표본이 남습니다. 따라서 필터를 평가할 때 정지 상태의 분산과 변화에 반응하는 시간을 함께 확인합니다. 이번 목표는 같은 입력에서 창 크기별 결과를 계산하고 가장 작은 분산만으로 창을 선택하지 않는 판단을 만드는 것입니다.

현재와 과거만 사용하는 창을 만듭니다

window가 3이면 현재 표본과 직전 두 표본을 평균합니다. 인덱스 i에서 시작은 max(0,i-window+1), 끝은 i+1입니다. 미래 표본은 쓰지 않으므로 온라인 제어에서 계산할 수 있는 인과적 필터입니다. 중심을 기준으로 앞뒤를 평균하는 방식은 오프라인 그림을 부드럽게 만들 수 있지만 아직 도착하지 않은 값을 필요로 하므로 이번 실습의 계약에 맞지 않습니다.

처음 window-1개 표본에서는 존재하는 표본만 평균합니다. window=3일 때 첫 출력은 x0, 둘째는 (x0+x1)/2입니다. 부족한 값을 0으로 채우면 센서가 낮은 거리를 읽은 것처럼 시작하고, 첫 값을 반복해서 채우는 방식은 다른 초기화 정책입니다. 어떤 방식도 이름만 이동평균이라고 적어 두면 결과가 달라질 수 있으므로 이번 실습에서는 부분 창 정책을 명시하고 전체 창 구간만 분산 평가에 사용합니다.

window는 1 이상의 정수이고 values 길이를 넘지 않습니다. window=1이면 원시 입력이 그대로 나옵니다. true는 Python에서 정수처럼 동작할 수 있지만 설정 값으로 받아들이지 않도록 type 검사를 사용합니다. 빈 입력과 비정상 표본은 ERROR입니다. 평균에서 null을 자동으로 빼면 창의 실제 기간이 들쭉날쭉해지므로 누락 정책 없이 값을 제거하지 않습니다.

정지 기록의 분산을 비교합니다

원시 분산은 입력의 평균에서 벗어난 값의 제곱 평균입니다. 필터 분산은 첫 window-1개 출력을 제외하고 나머지 출력의 평균을 기준으로 계산합니다. 첫 구간을 제외하는 이유는 초기 부분 창과 가득 찬 창을 섞지 않기 위해서입니다. 창마다 평가 표본 수는 다릅니다. 짧은 기록에서 계산한 분산은 관측 결과이며 센서 전체 분포의 정밀한 추정이라고 말하지 않습니다.

values=[1.1,0.9,1.1,0.9,1.1,0.9]는 검산하기 쉬운 고정 fixture입니다. 원시 분산은 0.01m²입니다. window=2의 가득 찬 평균은 모두 1.0이라 분산이 0이고 window=3은 1.0333과 0.9667 부근을 오갑니다. 창이 더 커졌는데 이 짧은 규칙 입력에서는 분산이 다시 커집니다. 짝수 창이 교대 패턴을 정확히 상쇄하기 때문이며 큰 창이면 모든 기록에서 더 좋아진다는 주장을 경계합니다.

독립이고 같은 분산인 잡음의 평균은 창 크기가 커질수록 분산이 줄어드는 이론적 성질이 있습니다. 실제 입력은 상관이 있거나 편향이 남을 수 있고 평가 기록 길이도 한정되어 있습니다. 이 레슨의 고정 교대 입력을 독립 가우시안 표본으로 해석하지 않습니다. 필터 평균이 안정되어도 공통 편향은 그대로 남기 때문에 앞 레슨의 bias와 RMSE도 같이 보는 것이 좋습니다.

계단 응답의 시간 원점을 고정합니다

지연 실험은 먼 거리 1.0m를 window개 넣은 뒤 가까운 거리 0.0m를 window개 넣습니다. 실제 센서에서 유효한 0m라는 뜻은 아니며 필터의 수학적 계단 시험 신호입니다. 정상 스캔의 최소 거리 규칙과 혼합하지 않습니다. 가까운 첫 표본을 시간 0으로 잡고 필터 출력이 0.25m 이하가 된 첫 인덱스에 dt_s를 곱해 crossing_s를 계산합니다.

window=4에서는 가까워진 첫 표본의 평균이 0.75, 다음이 0.50, 그다음이 0.25입니다. 첫 표본을 인덱스 0으로 잡았으므로 dt=0.05초에서 crossing_s는 0.10초입니다. 표본 개수 3에 dt를 곱한 0.15초가 아닙니다. 코드와 보고서에서 변화가 발생한 시각과 확인한 시각의 차이를 사용하는지 확인합니다. window=1이면 변화 첫 표본부터 조건을 만족해 0.0초입니다.

문턱을 정확히 포함하므로 비교는 0.25 이하입니다. 미만을 쓰면 window=4에서는 한 tick 더 기다리게 됩니다. 이번 실습의 지연은 특정 문턱의 계단 응답이며 선형 필터의 모든 주파수에서 같은 지연이 생긴다는 뜻은 아닙니다. 필터 전달 특성과 그룹 지연의 자세한 이론 대신 지금 구현한 정지 판단이 몇 번의 표본 뒤에 일어나는지 직접 검사합니다.

통신 지연과 필터 지연을 구별합니다

stamp와 received의 차이는 전달 또는 처리 경로에서 생기는 지연입니다. 창에 남은 과거 값 때문에 반응이 느린 현상은 필터 지연입니다. 측정 시각을 정확히 복구해도 여러 시각의 값을 평균한 결과는 단일 현재 거리와 같아지지 않습니다. 반대로 필터 창을 줄여도 메시지가 늦게 도착하는 문제는 남습니다. 두 원인을 따로 기록해야 수정 대상이 설정인지 통신 경로인지 판단할 수 있습니다.

로봇이 0.2m/s로 움직일 때 0.10초를 더 기다리면 일정 속도 가정에서 0.02m를 추가로 이동합니다. 실제 정지에는 제어 주기와 감속·미끄러짐도 영향을 줍니다. 이 숫자는 필터만의 추가 이동량 계산이며 전체 제동 거리가 아닙니다. 창을 선택할 때 분산 감소와 이 거리, 지도 여유, 입력이 끊긴 경우를 함께 고려합니다. 작은 분산 하나만으로 안전하다는 결론을 내리지 않습니다.

안전 정지 입력과 화면용 부드러운 표시를 분리할 수도 있습니다. 표시에는 평균을 쓰더라도 오래된 입력이나 아주 가까운 원시 거리는 즉시 거절하는 정책을 둘 수 있습니다. 이번 미션의 RangeGate는 이동평균을 안전 경로에 넣지 않고 유효 범위와 시각·근접을 검사합니다. 필터를 추가하려면 이 정책을 유지하며 어느 신호에 적용했는지, 추가 지연은 얼마인지 증거를 남깁니다.

틀린 창 계산을 출력으로 찾습니다

슬라이스 끝 인덱스를 i로 쓰면 현재 표본이 제외되고 window=1에서 빈 창을 만들 수 있습니다. 시작 인덱스에 +1을 빼먹으면 최대 window+1개가 들어옵니다. 고정 입력의 출력 목록을 확인하고 첫 세 값과 충분히 찬 창의 개수를 손으로 세어 보세요. ZeroDivisionError는 빈 chunk를 만든 경우, StopIteration은 계단 기록 안에서 문턱을 찾지 못한 경우부터 의심합니다.

출력은 [원시 분산,[[창 크기,필터 분산,crossing_s],...]]입니다. 분산은 m², crossing_s는 초라서 두 숫자의 크기를 직접 비교하지 않습니다. 창별로 한 행을 적고 필요한 정지 응답 시간을 만족하는 후보 안에서 신호 안정성을 판단합니다. 결과는 여섯 자리 수준으로 반올림하며 계산 중에는 원래 실수를 유지합니다. 입력 순서와 출력 행 순서는 같으므로 같은 창의 지표를 묶어 읽습니다.

실습을 마친 뒤 교대 입력·상수 입력·한 표본·큰 창·잘못된 dt를 비교합니다. 상수 입력의 분산은 0이지만 계단 응답은 창에 따라 다릅니다. 이는 정지 상태의 평가만으로 변화 대응을 검증할 수 없다는 근거입니다. 더 읽기에서는 지수 이동평균과 필터 선택의 배경을 이어 읽습니다. 이번 제출에는 분산과 crossing_s를 함께 근거로 들어 한 창을 선택하는 문장을 추가합니다.

따라하기

부분 창의 시작을 확인합니다

창이 채워지기 전에도 현재까지 존재하는 값만 평균합니다.

import json,sys,math

def moving(values,window):
    out=[]
    for i in range(len(values)):
        chunk=values[max(0,i-window+1):i+1]
        out.append(sum(chunk)/len(chunk))
    return out

def evaluate(obj):
    values=obj['values'];windows=obj['windows'];dt=obj['dt_s']
    if not isinstance(values,list) or not values or any(type(v) not in (int,float) or not math.isfinite(v) for v in values): raise ValueError('VALUES')
    if not isinstance(windows,list) or not windows or any(type(w) is not int or w<1 or w>len(values) for w in windows): raise ValueError('WINDOW')
    if type(dt) not in (int,float) or not math.isfinite(dt) or dt<=0: raise ValueError('DT')
    mean=sum(values)/len(values);raw=sum((v-mean)**2 for v in values)/len(values)
    results=[]
    for window in windows:
        filtered=moving(values,window)[window-1:]
        avg=sum(filtered)/len(filtered);variance=sum((v-avg)**2 for v in filtered)/len(filtered)
        step=moving([1.0]*window+[0.0]*window,window)[window:]
        crossing=next(i for i,v in enumerate(step) if v<=.25)
        results.append([window,round(variance,6),round(crossing*dt,6)])
    return [round(raw,6),results]

print(moving([1.,3.,5.,7.],3))

실행 결과

[1.0, 2.0, 3.0, 5.0]

창별 분산과 문턱 지연을 비교합니다

각 행의 마지막 값은 초, 가운데 값은 m²입니다. 교대 기록의 창 2와 3을 비교합니다.

import json,sys,math

def moving(values,window):
    out=[]
    for i in range(len(values)):
        chunk=values[max(0,i-window+1):i+1]
        out.append(sum(chunk)/len(chunk))
    return out

def evaluate(obj):
    values=obj['values'];windows=obj['windows'];dt=obj['dt_s']
    if not isinstance(values,list) or not values or any(type(v) not in (int,float) or not math.isfinite(v) for v in values): raise ValueError('VALUES')
    if not isinstance(windows,list) or not windows or any(type(w) is not int or w<1 or w>len(values) for w in windows): raise ValueError('WINDOW')
    if type(dt) not in (int,float) or not math.isfinite(dt) or dt<=0: raise ValueError('DT')
    mean=sum(values)/len(values);raw=sum((v-mean)**2 for v in values)/len(values)
    results=[]
    for window in windows:
        filtered=moving(values,window)[window-1:]
        avg=sum(filtered)/len(filtered);variance=sum((v-avg)**2 for v in filtered)/len(filtered)
        step=moving([1.0]*window+[0.0]*window,window)[window:]
        crossing=next(i for i,v in enumerate(step) if v<=.25)
        results.append([window,round(variance,6),round(crossing*dt,6)])
    return [round(raw,6),results]

print(evaluate(dict(values=[1.1,.9,1.1,.9,1.1,.9],windows=[1,2,3,4],dt_s=.05)))

실행 결과

[0.01, [[1, 0.01, 0.0], [2, 0.0, 0.05], [3, 0.001111, 0.1], [4, 0.0, 0.1]]]

계단의 시간 원점을 검산합니다

첫 가까운 입력을 0초로 놓습니다. 추가 이동량은 전체 제동 거리가 아닙니다.

import json,sys,math

def moving(values,window):
    out=[]
    for i in range(len(values)):
        chunk=values[max(0,i-window+1):i+1]
        out.append(sum(chunk)/len(chunk))
    return out

def evaluate(obj):
    values=obj['values'];windows=obj['windows'];dt=obj['dt_s']
    if not isinstance(values,list) or not values or any(type(v) not in (int,float) or not math.isfinite(v) for v in values): raise ValueError('VALUES')
    if not isinstance(windows,list) or not windows or any(type(w) is not int or w<1 or w>len(values) for w in windows): raise ValueError('WINDOW')
    if type(dt) not in (int,float) or not math.isfinite(dt) or dt<=0: raise ValueError('DT')
    mean=sum(values)/len(values);raw=sum((v-mean)**2 for v in values)/len(values)
    results=[]
    for window in windows:
        filtered=moving(values,window)[window-1:]
        avg=sum(filtered)/len(filtered);variance=sum((v-avg)**2 for v in filtered)/len(filtered)
        step=moving([1.0]*window+[0.0]*window,window)[window:]
        crossing=next(i for i,v in enumerate(step) if v<=.25)
        results.append([window,round(variance,6),round(crossing*dt,6)])
    return [round(raw,6),results]

s=moving([1.]*4+[0.]*4,4)[4:]
print(s)
t=next(i for i,v in enumerate(s) if v<=.25)*.05
print(f'crossing_s={t:.2f} extra_m={.2*t:.3f}')

실행 결과

[0.75, 0.5, 0.25, 0.0]
crossing_s=0.10 extra_m=0.020

확인 문제

실습

valuesは固定入力のm値、windowsは1以上かつ入力長以下の整数の一覧、dt_sは正の周期です。movingのTODOを現在と過去の部分窓の平均にします。出力は[原分散,[[window,満窓出力の分散,crossing_s],...]]です。分散の分母は評価件数、単位はm²です。各windowについて1.0をwindow個・0.0をwindow個流し、最初の0.0入力を時刻0として出力が0.25以下になる最初の時刻を測ります。無効入力はERRORです。

모범 답안
import json,sys,math

def moving(values,window):
    out=[]
    for i in range(len(values)):
        chunk=values[max(0,i-window+1):i+1]
        out.append(sum(chunk)/len(chunk))
    return out

def evaluate(obj):
    values=obj['values'];windows=obj['windows'];dt=obj['dt_s']
    if not isinstance(values,list) or not values or any(type(v) not in (int,float) or not math.isfinite(v) for v in values): raise ValueError('VALUES')
    if not isinstance(windows,list) or not windows or any(type(w) is not int or w<1 or w>len(values) for w in windows): raise ValueError('WINDOW')
    if type(dt) not in (int,float) or not math.isfinite(dt) or dt<=0: raise ValueError('DT')
    mean=sum(values)/len(values);raw=sum((v-mean)**2 for v in values)/len(values)
    results=[]
    for window in windows:
        filtered=moving(values,window)[window-1:]
        avg=sum(filtered)/len(filtered);variance=sum((v-avg)**2 for v in filtered)/len(filtered)
        step=moving([1.0]*window+[0.0]*window,window)[window:]
        crossing=next(i for i,v in enumerate(step) if v<=.25)
        results.append([window,round(variance,6),round(crossing*dt,6)])
    return [round(raw,6),results]
try: print(json.dumps(evaluate(json.load(sys.stdin))))
except (ValueError,TypeError,KeyError,OverflowError): print('ERROR')

더 읽기

면접 질문

  • 센서 잡음을 추가한 시뮬레이션을 어떻게 평가하는지 설명해 주시면 됩니다.