그림 일치도로 SAX 최적 변수 찾기

일전에는 SAX 기법을 이용하여 데이터를 정제하는 기법에 대해서 알아 보았습니다. 

이번에는 어떻게 하면 SAX에 값을 세팅해서 최적으로 이상치를 제거할 수 있을지 시도해봤습니다. 

일전 SAX 포스팅은 다음 버튼을 통해 확인해주세요. 

그냥 SAX 로 시도했을 때는?

이상치를 제거하고 싶은 원본 데이터입니다. 

이런 원본 데이터에서 일반적인 SAX 기법을 이용하면 어떤 결과를 얻을 수 있을까요?

SAX 기법을 통해 특정 Labeling에 해당하는 값만 도출했을 때의 결과입니다.

아주 튀는 값들은 잘 없어졌지만, 보다 더 깨끗하게 데이터를 정리하고 싶습니다.

이를 위해서 이미지 일치도를 이용해보면 어떨가 싶었습니다. 

이미지 일치도 비교

저는 직접 원래 원본에 있던 데이터를 다음과 같이 깎았습니다.

위 아래로 삐죽삐죽 튀어나온 데이터를 제거한 것이죠.

이런 이미지를 정답이라고 가정을 하겠습니다. 

그리고 위 그림은 조금 듬성듬성 원본 데이터에서 이상치를 제거한 그림입니다.

뒷부분에 삐죽삐죽 이상치가 보이죠?

SAX 기법을 통해 이상치를 제거한 그림이 바로 위의 그림과 같다고 가정하겠습니다. 

위 그림이 제출한 답변입니다.

그리고 이 답변을 아까 제가 위에서 손수 다 깎은 정답과 비교를 하는 겁니다. 

정답과 답변의 일치도를 비교하는 것이지요.

일치도가 높으면 높을 수록 더 완벽한 SAX 변수가 적용되었다는 뜻이고

이를 통해서 어떤 값으로 SAX 세팅을 하면 이상치 제거에 능할지 알 수 있지 않을까?

하는 생각입니다. 

여기서 사용한 방식은 SSIM 입니다.

코드는 다음과 같이 나타낼 수 있겠습니다. 

 

				
					from skimage.metrics import structural_similarity as ssim
from PIL import Image
import numpy as np
import matplotlib.pyplot as plt

# 이미지 파일 경로 설정
img1_path = "gt.png"      # 첫 번째 이미지 파일 경로
img2_path = "answer.png"  # 두 번째 이미지 파일 경로

# 두 이미지 파일을 로드하고 grayscale로 변환
img1 = Image.open(img1_path).convert("L")
img2 = Image.open(img2_path).convert("L")

# 이미지 크기를 동일하게 조정 (크기가 다를 경우)
img1 = img1.resize(img2.size)

# numpy 배열로 변환
img1_array = np.array(img1)
img2_array = np.array(img2)

# SSIM 계산 및 차이 맵 생성
similarity_index, diff = ssim(img1_array, img2_array, full=True)

# 차이 맵을 0~255 사이의 값으로 변환하여 시각화 가능하게 변환
diff = (diff * 255).astype("uint8")

# 시각화
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
axes[0].imshow(img1_array, cmap="gray")
axes[0].set_title("Ground Truth Image")
axes[0].axis("off")

axes[1].imshow(img2_array, cmap="gray")
axes[1].set_title("Answer Image")
axes[1].axis("off")

axes[2].imshow(diff, cmap="hot")
axes[2].set_title("SSIM Difference Map")
axes[2].axis("off")

plt.suptitle(f"SSIM Index: {similarity_index:.4f}", fontsize=16)
plt.show()

				
			

위 코드를 이용하면 이미지의 일치도를 비교하는 것이 가능한 것이죠.

최대한 값이 1에 가까울 수록 두 이미지가 일치한다는 뜻입니다.

SAX 와 SSIM 기법을 동시에 사용하기

 

SAX 기법을 이용하여 특정한 범위에 해당하는 데이터들을 솎아냅니다.

이를 이용하여 그래프를 그리고, 그린 그래프는 SSIM 기법을 이용하여 정답과 비교합니다. 

이 과정을 여러 번 반복하여 가장 일치도가 높은 SAX 세팅값을 찾아낼 겁니다.

이를 위해 우선은 다음과 같이 1회 반복을 위한 코드를 짜겠습니다. 

				
					import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
from skimage.metrics import structural_similarity as ssim
from scipy.stats import norm

# 테스트를 위한 데이터 로드
path = r"VitalStudy\Data\TestSetV2.csv"
df = pd.read_csv(path)

# 분석을 하고자 하는 시계열 데이터의 데이터입니다.
data = df["data"].tolist()

# 데이터의 길이를 N으로 지정합니다.
N = len(data)

# 데이터를 정규화합니다.
NormData = []
Xmean = np.nanmean(data)
std_dev = np.nanstd(data)

# 각각의 데이터 포인트를 Z-정규화
for x in data:
    Xnorm = (x - Xmean) / std_dev
    NormData.append(Xnorm)
    
# 전체 데이터를 몇 등분하면 좋을 지... 
R = 10

# PAA 적용을 위한 구간 개수 W를 설정합니다.
W = N // R

# PAA 변환: 데이터를 W 개의 구간으로 나누어 각 구간의 평균을 계산
paa_values = []
segment_size = N / W

for i in range(W):
    # 구간의 시작과 끝 인덱스를 계산합니다.
    start_idx = int(i * segment_size)
    end_idx = int((i + 1) * segment_size) if i < W - 1 else N  # 마지막 구간은 끝까지 포함
    
    # 구간 내 평균값 계산
    segment_sum = np.sum(NormData[start_idx:end_idx])
    segment_value = (W / N) * segment_sum
    paa_values.append(segment_value)

# PAA 데이터를 원본 데이터의 길이와 일치시키기 위해 각 구간을 평균값으로 대체함
paa_data_expanded = np.repeat(paa_values, np.ceil(segment_size).astype(int))[:N]

# 부호화 단계
alphabet = ["a", "b", "c"]

# breakpoints를 특정 숫자 두 개로 설정합니다. 0~1 사이의 2개의 수를 선택합니다. 
# x1 은 x2 보다 작은 수로 설정합니다. 
x1 = 0.25
x2 =  0.75
breakpoints = norm.ppf([x1, x2])

# SAX 변환
sax_symbols = []
for value in paa_values:
    if value < breakpoints[0]:
        sax_symbols.append("a")
    elif breakpoints[0] <= value <= breakpoints[1]:
        sax_symbols.append("b")  # 구간 내 값에 할당할 기호
    else:
        sax_symbols.append("c")

# 남기고 싶은 부호를 설정합니다.
valid_alphabet = ["b"]

# 원본 데이터에서 유효 기호에 해당하지 않는 값을 NaN으로 변경
filtered_original_data = data.copy()
for i, symbol in enumerate(sax_symbols):
    start_idx = int(i * segment_size)
    end_idx = int((i + 1) * segment_size) if i < W - 1 else N
    if symbol not in valid_alphabet:
        filtered_original_data[start_idx:end_idx] = [np.nan] * (end_idx - start_idx)

# 필터링된 데이터를 파일로 저장
filtered_df = pd.DataFrame({"Original": data, "Filtered": filtered_original_data})
filtered_df.to_csv("filtered_data.csv", index=False)

# gt 이미지의 크기를 가져옵니다
gt_image_path = r"VitalStudy\Data\Image\gt.png"
goal = r"VitalStudy\Data\Image\answer_graph.png"

gt_image = Image.open(gt_image_path)
width, height = gt_image.size

# GT 이미지의 DPI를 정수로 가져옴
dpi = gt_image.info.get('dpi', (72,))[0]  # 첫 번째 값만 가져오도록 수정

# X, Y축 범위를 설정 (GT 이미지와 동일하게 약간 여유 추가)
x = np.linspace(0, 1, N)
plt.plot(x, filtered_original_data, c="k")

# X축을 양쪽으로 5% 여유를 두어 설정
plt.xlim(-0.05, 1.05)

# Y축을 데이터의 최소/최대값에서 10% 여유를 두어 설정
y_min, y_max = np.nanmin(data), np.nanmax(data)
y_range = y_max - y_min
plt.ylim(y_min - 0.1 * y_range, y_max + 0.1 * y_range)

# 저장할 때도 DPI를 설정해 원본과 동일한 크기로 저장
plt.savefig(goal, pad_inches=0, dpi=dpi)
plt.close()


# 저장한 그림과 gt의 그림을 가지고 일치도를 조사한다.
# 이미지 파일 경로
answer_path = goal
gt_image_path = gt_image_path  # 두 번째 이미지 파일 경로

# 두 이미지 파일을 로드하고 grayscale로 변환
gt = Image.open(gt_image_path).convert("L")
answer = Image.open(answer_path).convert("L")

# 이미지 크기를 동일하게 조정 (크기가 다를 경우)
gt = gt.resize(answer.size)

# numpy 배열로 변환
gt_array = np.array(gt)
answer_array = np.array(answer)

# SSIM 계산 및 차이 맵 생성
similarity_index, diff = ssim(gt_array, answer_array, full=True)

# 일치도를 출력한다
print(similarity_index)





				
			

위와 같이 코드를 짰습니다.

이 경우 특정한 R, X1, X2 라는 값에 의해서 필터링된 이미지를 생성하고

해당 이미지가 얼마나 정답과 가까운지 결과를 출력하는 코드입니다.

한 번 돌려볼까요?

				
					0.9249722862548323
				
			

위 코드에서는 그냥 이렇게 숫자가 딱 나타납니다.

제 목적은 이 숫자를 한없이 1에 가깝도록 할 겁니다. 

위 작업을 반복하며 최적의 값을 찾아보자

				
					import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
from skimage.metrics import structural_similarity as ssim
from scipy.stats import norm
from skopt import gp_minimize
from skopt.space import Integer, Real
from skopt.utils import use_named_args

# 데이터 로드 및 전처리
path = r"VitalStudy\Data\TestSetV2.csv"
df = pd.read_csv(path)
data = df["data"].tolist()
N = len(data)

# GT 이미지 경로 및 설정
gt_image_path = r"VitalStudy\Data\Image\gt.png"
goal = r"VitalStudy\Data\Image\answer_graph.png"
gt_image = Image.open(gt_image_path)
width, height = gt_image.size
dpi = gt_image.info.get('dpi', (72,))[0]

# 검색할 파라미터 공간 정의
space  = [
    Integer(10, 100, name='R'),  # R 범위
    Real(0.25, 0.5, name='X1'),    # X1 범위
    Real(0.5, 0.9, name='X2')      # X2 범위
]

# 목적 함수 정의
@use_named_args(space)
def objective(R, X1, X2):
    if X1 >= X2:
        return 1.0  # 유효하지 않은 조합에 높은 손실 할당

    # 데이터 정규화
    NormData = [(x - np.nanmean(data)) / np.nanstd(data) for x in data]

    # PAA 적용
    W = N // R
    paa_values = [np.mean(NormData[int(i * N / W):int((i + 1) * N / W)]) for i in range(W)]
    paa_data_expanded = np.repeat(paa_values, int(np.ceil(N / W)))[:N]

    # SAX 변환 및 필터링
    breakpoints = norm.ppf([X1, X2])
    sax_symbols = ["a" if val < breakpoints[0] else "b" if breakpoints[0] <= val <= breakpoints[1] else "c" for val in paa_values]
    
    # 필터링된 데이터 생성 (paa_data_expanded와 동일한 길이로 확장)
    filtered_original_data = []
    for val, sym in zip(paa_data_expanded, np.repeat(sax_symbols, int(np.ceil(N / W)))[:N]):
        if sym == "b":
            filtered_original_data.append(val)
        else:
            filtered_original_data.append(np.nan)

    # 시각화 및 이미지 저장
    x = np.linspace(0, 1, N)
    plt.plot(x, filtered_original_data, c="k")
    plt.xlim(-0.05, 1.05)
    y_min, y_max = np.nanmin(data), np.nanmax(data)
    plt.ylim(y_min - 0.1 * (y_max - y_min), y_max + 0.1 * (y_max - y_min))
    plt.savefig(goal, pad_inches=0, dpi=dpi)
    plt.close()

    # SSIM 계산
    gt = Image.open(gt_image_path).convert("L").resize((width, height))
    answer = Image.open(goal).convert("L").resize((width, height))
    similarity_index, _ = ssim(np.array(gt), np.array(answer), full=True)

    # SSIM의 -1을 반환 (최대화 문제를 최소화 문제로 변환)
    return -similarity_index

# 베이즈 최적화 수행
res = gp_minimize(objective, space, n_calls=10, random_state=0)  # 30번의 최적화 시도

# 최적의 파라미터 및 결과 출력
best_R, best_X1, best_X2 = res.x
best_similarity = -res.fun  # objective 함수에서 음수로 반환했으므로, 다시 양수로 변환

print("Best Results:")
print(f"Best Similarity: {best_similarity}")
print(f"Best R: {best_R}, Best X1: {best_X1}, Best X2: {best_X2}")

				
			

베이즈 최적화 기법을 이용하여 최적의 일치도 값을 찾기 위해 반복합니다.

gp_minimize() 함수에서 n_calls 를 크게 설정할 수록 최적화 시도 숫자가 늘어납니다. 

끝에서 최적의 파리미터 및 결과를 출력할 것이고 

이것이 모든 시도에 있어서 가장 일치도가 높은 그림이라 보면 되겠습니다. 

마치며...

 

데이터의 이상치 제거를 위해 또 다른 방법론이나 아이디어를 많이 사용을 해봐야겠습니다.

해당 방법도 썩 만족할 만큼 좋지는 못하네요.ㅎㅎ