센서 데이터가 서로 다른 길이를 갖고 있고, 일부는 센서 고장으로 0으로 끊긴 경우까지 고려한 보간 + 거리 계산 프로그램입니다. 이 코드는 다음과 같은 구조로 설계되어 있습니다:
1. 목표**
- 길이가 다른 센서 데이터를 **동일한 길이로 보간(interpolate)
- 센서 중 일부는 고장으로 **데이터가 조기 종료**되는 경우 존재
- 보간 후 **센서 간 유클리디안 거리 계산**
2. 핵심 설계 포인트
- **0값으로 센서가 멈췄을 경우**, 그 시점까지만 유효 데이터로 판단하여 보간
- 센서 간 거리는 `scipy.spatial.distance.euclidean`으로 계산
- 센서별로 하나의 기준 길이(`target_length`)로 보간 처리
3. 주요 함수 설명
- 센서 데이터에서 유효 값(0이 아닌 구간)만 추출
- 마지막 유효값까지만 사용해서 선형 보간
- 고장 등으로 유효값이 2개 미만일 경우, **전부 0으로 채움**
4. 출력
- 센서 간의 거리 행렬(`distance_matrix`)을 `pandas.DataFrame`으로 출력
- 예: `sensor_A`와 `sensor_B` 사이의 거리 확인 가능
---------------------
import numpy as np
import pandas as pd
from scipy.interpolate import interp1d
from scipy.spatial.distance import euclidean
# 시계열 데이터를 기준 길이로 보간하는 함수
def interpolate_series(series, target_length):
non_zero_indices = np.nonzero(series)[0]
# 모든 값이 0이거나 센서가 너무 일찍 멈춘 경우 처리
if len(non_zero_indices) < 2:
return np.zeros(target_length)
valid_end = non_zero_indices[-1] + 1
truncated = series[:valid_end]
x_old = np.linspace(0, 1, num=len(truncated))
f = interp1d(x_old, truncated, kind='linear')
x_new = np.linspace(0, 1, num=target_length)
interpolated = f(x_new)
return interpolated
# 두 센서 데이터 간 거리 계산 함수
def calculate_distance(series1, series2, target_length=100):
interp1 = interpolate_series(series1, target_length)
interp2 = interpolate_series(series2, target_length)
return euclidean(interp1, interp2)
# 예제: 다양한 길이와 중간에 0으로 끊긴 데이터
sensor_data = {
'sensor_A': np.array([10, 15, 13, 11, 9, 6, 3, 0, 0, 0]), # 멈춘 센서
'sensor_B': np.array([8, 12, 14, 13, 10, 8, 6, 4, 2, 1]), # 정상 센서
'sensor_C': np.array([5, 7, 6, 0, 0, 0, 0, 0, 0, 0]) # 일찍 멈춤
}
# 기준 길이
target_len = 50
# 센서 간 거리 계산
distance_matrix = pd.DataFrame(index=sensor_data.keys(), columns=sensor_data.keys())
for name1 in sensor_data:
for name2 in sensor_data:
dist = calculate_distance(sensor_data[name1], sensor_data[name2], target_len)
distance_matrix.loc[name1, name2] = dist
print("센서 간 거리 행렬:")
print(distance_matrix)
