14주차 · 통합 코드 워크스루

학습 목표

W2–W13에서 배운 개념이 하나의 파이프라인 안에서 어떻게 연결되는지 읽는다
NumPy로 만든 센서 시뮬레이션 데이터가 통계 처리로 이어지는 흐름을 추적한다
함수와 NumPy mask가 불량 데이터를 걸러내는 위치를 확인한다
f-string 테이블과 요약 리포트가 어떤 값을 보여 주는지 해석한다
그래프 설계안(plot_spec)의 제목·축·단위·데이터 길이를 점검한다
파이프라인 각 단계에서 assert가 무엇을 보호하는지 설명한다

워크스루 개요: 센서 데이터 분석 파이프라인

시나리오: 건물 내부 온도 센서 1개가 1분 간격으로 24회 측정 데이터를 기록했습니다. 일부 시점에는 불량 데이터(음수값, 극단값)가 섞일 수 있습니다. 이 단일 시계열 데이터를 정제하고, 통계를 계산하고, 요약 리포트를 출력한 뒤 시각화 설계안을 작성합니다.

실제 코드는 한 가지 문법만 쓰는 일이 거의 없습니다. 위 시나리오 하나를 처리하는 데도 NumPy 배열(W11), 함수 분해(W5·W9), 자료구조(W6), 카운팅(W13), 시각화 설계(W12)가 동시에 등장합니다. 그래서 코드가 길어지면 “한 줄씩 무슨 문법인가”보다 **“지금 데이터가 어느 단계에 있고, 다음 단계로 무엇이 넘어가는가”**를 따라 읽어야 합니다. 이 워크스루로 그 흐름 읽기를 연습합니다.

파이프라인 5단계:

단계	작업	핵심 개념
Step 1	데이터 생성	NumPy 배열, `np.random` 패턴 (W12 복습)
Step 2	처리 함수	필터링, 통계, 분류 함수 (W5, W9, W11)
Step 3	검증	범위 조건, NumPy mask, shape 점검 (W10–W11)
Step 4	출력	f-string 테이블, 요약 리포트 (W5)
Step 5	시각화 설계안	`plot_spec` 딕셔너리 (W12)

센서 데이터 분석 파이프라인의 5단계 데이터 흐름

위 그림은 데이터가 다섯 단계를 거치며 어떻게 변하는지 보여 줍니다. 큰 변화는 Step 3에서 생깁니다. 불량값이 제거되면서 배열 길이가 (24,)에서 (22,)로 줄어듭니다. 코드를 읽을 때 “이 변수의 길이가 지금 24인가 22인가”를 항상 의식하면, 어디서 데이터가 걸러졌는지 추적할 수 있습니다.

Step 1 · 데이터 생성

안내

파이프라인을 만들거나 읽을 때, 가장 먼저 필요한 것은 언제든 똑같이 재현되는 입력 데이터입니다. 12주차 시각화 예제에서 사용한 np.random.default_rng(42) 패턴이 그 역할을 합니다. 괄호 안의 숫자(seed)를 고정하면 매번 같은 난수가 나오므로, 코드를 다시 실행해도 결과가 바뀌지 않습니다. 디버깅이나 채점에서 “내 화면과 네 화면이 다르다”는 혼란을 막아 주는 장치입니다.

실제 센서처럼 기저 온도에 노이즈를 더하고, 여기에 의도적으로 불량값(음수, 극단값)을 심어 둡니다. 정상 데이터만으로는 “거르는 코드”가 제대로 동작하는지 확인할 수 없기 때문입니다. 즉 Step 1은 데이터를 만들면서 뒤 단계의 검증 로직을 시험할 함정도 함께 심어 두는 단계입니다. np.arange(n)은 0, 1, 2, ..., n-1 형태의 측정 인덱스를 만들어, 나중에 “몇 번째 측정인지”를 추적하는 기준이 됩니다.

# 참고: 데이터 생성 패턴
import numpy as np

rng = np.random.default_rng(42)           # 재현 가능한 난수 생성기
n = 24                                     # 측정 횟수
raw_time = np.arange(n)                    # 원래 측정 인덱스 (0~23)
base_temp = 22.0                           # 기저 온도 (°C)
temp = base_temp + rng.normal(scale=1.5, size=n)   # 노이즈 추가

불량 데이터 주입: 실제 센서 오류를 모사하려면 몇 개의 값을 의도적으로 음수나 극단값으로 바꿉니다.

temp[3] = -5.0    # 오류 데이터 (음수)
temp[17] = 80.0   # 오류 데이터 (극단값)

Step 2 · 처리 함수

안내

이 단계의 원칙은 9주차에서 배운 “함수 하나는 한 가지 일만”(단일 책임)입니다. 한 함수에 필터링·통계·분류를 모두 넣으면, 어디서 문제가 생겼는지 찾기 어렵고 재사용도 안 됩니다. 그래서 역할별로 make_valid_mask(거를 위치 표시), filter_valid(거르기), compute_stats(통계), classify_temp(분류)로 나눕니다.

이렇게 나누면 세 가지 이점이 생깁니다. 첫째, 각 함수를 따로 테스트할 수 있습니다. 둘째, 같은 함수를 다른 데이터에 재사용할 수 있습니다. 셋째, 코드를 읽을 때 함수 이름만 봐도 “이 줄이 무슨 일을 하는지” 알 수 있어 전체 흐름이 문장처럼 읽힙니다. 워크스루에서 코드를 추적할 때도, 함수 본문보다 “어떤 함수가 어떤 순서로 호출되는가”를 먼저 보세요.

# 참고: 처리 함수 패턴
import numpy as np

def make_valid_mask(data, low, high):
    """유효 범위 [low, high] 안이면 True, 아니면 False인 mask를 만듭니다."""
    return (data >= low) & (data <= high)

def filter_valid(data, valid_mask):
    """valid_mask가 True인 위치의 값만 남깁니다."""
    return data[valid_mask]

def compute_stats(data):
    """mean, std, min, max 를 딕셔너리로 반환합니다."""
    assert len(data) > 0, "통계를 계산할 정상 데이터가 없습니다"
    return {
        "mean": round(float(data.mean()), 2),
        "std":  round(float(data.std()),  2),
        "min":  round(float(data.min()),  2),
        "max":  round(float(data.max()),  2),
    }

def classify_temp(temp_c):
    """온도를 3단계로 분류합니다."""
    if temp_c < 18:
        return "cold"
    elif temp_c <= 26:
        return "normal"
    else:
        return "hot"

Step 3 · 검증

안내

불량 데이터를 거를 때는 먼저 정상 범위를 조건으로 정하고, NumPy boolean mask로 정상 데이터만 남깁니다. boolean mask는 [True, False, True, ...]처럼 각 위치를 남길지(True) 버릴지(False) 표시하는 배열입니다. (data >= 0) & (data <= 50)처럼 조건을 쓰면, NumPy가 모든 위치에 대해 한 번에 True/False를 계산해 mask를 만들어 줍니다.

같은 mask를 온도와 시간 인덱스에 함께 적용하는 모습

이 단계의 핵심 개념은 **“같은 mask를 두 배열에 함께 적용한다”**입니다. 위 그림처럼 불량값(인덱스 3·5)을 빼면 온도 배열은 24개에서 22개로 줄어듭니다. 이때 시간 인덱스 배열에도 똑같은 mask를 적용하면, 남은 온도값이 각각 “몇 번째 측정”이었는지가 그대로 보존됩니다. 만약 온도만 거르고 시간 인덱스를 그대로 두면 22개의 온도와 24개의 시간이 어긋나 그래프의 x축과 y축이 맞지 않습니다. 그래서 clean_temp = raw_temp[mask]와 clean_time = raw_time[mask]를 항상 짝으로 적용합니다.

또한 이 워크스루에서는 정제 후 데이터가 비어 있는지 명시적으로 확인하고, 비어 있으면 평균 대신 None을 반환합니다. 모든 값이 불량이어서 하나도 안 남는 극단적인 경우에 mean()을 부르면 오류가 나기 때문에, “빈 배열”이라는 실패 케이스를 미리 막아 두는 것입니다.

# 참고: 검증 패턴
def checked_mean(data):
    """데이터가 비어 있으면 None을 반환합니다."""
    if len(data) == 0:
        print("경고: 정상 범위 데이터가 없습니다")
        return None
    return float(data.mean())

Step 4 · 출력

안내

계산이 끝났다면, 사람이 읽고 판단할 수 있는 형태로 결과를 정리해야 합니다. 숫자가 줄줄이 쏟아지는 출력은 정보가 있어도 읽히지 않습니다. f-string의 포맷 지정자(W5)를 쓰면 자릿수를 맞추고 열을 정렬해, 표처럼 한눈에 비교되는 리포트를 만들 수 있습니다.

여기서 두 가지 포맷 도구를 씁니다. {val:>8.2f}는 “소수점 둘째 자리까지, 오른쪽 정렬, 너비 8칸”이라는 뜻이라 숫자의 소수점이 세로로 가지런히 맞습니다. {key:<8}은 “왼쪽 정렬, 너비 8칸”이라 항목 이름이 줄을 맞춥니다. 이렇게 정렬하면 값을 세로로 비교하기 쉬워져, 표가 그대로 의사결정 근거가 됩니다.

# 참고: 출력 패턴
stats = {"mean": 22.3, "std": 1.4, "min": 19.8, "max": 24.7}

print(f"{'항목':<8} {'값':>8}")
print("-" * 18)
for key, val in stats.items():
    print(f"{key:<8} {val:>8.2f}")

출력 예시:

항목          값
------------------
mean      22.30
std        1.40
min       19.80
max       24.70

Step 5 · 시각화 설계안

안내

마지막 단계는 12주차의 핵심 메시지를 잇습니다. 그림을 그리기 전에 무엇을 그릴지 먼저 글로 정한다는 원칙입니다. plot_spec 딕셔너리는 제목·축 라벨·단위·차트 종류·데이터를 한 곳에 모은 시각화 설계안입니다. 실제 렌더링은 로컬 Python(matplotlib)에서 하고, 브라우저에서는 이 설계안을 작성하고 self-check로 점검하는 데 집중합니다. 설계안이 먼저 있으면 어떤 도구로 그리든 일관된 그래프가 됩니다.

여기서 tolist()가 왜 필요한지도 보겠습니다. NumPy 배열은 계산에는 좋지만, plot_spec처럼 나중에 JSON으로 옮기거나 다른 도구에 넘길 설계안에는 평범한 Python 리스트가 더 다루기 쉽습니다. 그래서 clean_time.tolist()로 배열을 리스트로 바꿉니다. 마지막 self-check에서는 labels_ready(제목·축 라벨이 모두 채워졌는가)와 same_length(x와 y의 길이가 같은가)를 확인합니다. 둘 다 그래프가 어긋나는 흔한 실수라, 그리기 전에 미리 막아 줍니다.

# 참고: 시각화 설계안 패턴
import numpy as np

clean_time = np.array([0, 1, 2, 4, 5])      # 불량값 제거 후에도 원래 측정 인덱스를 보존
clean_temp = np.array([22.1, 22.3, 22.0, 22.4, 22.2])

plot_spec = {
    "title": "Sensor Temperature over 24 Readings",
    "x_label": "Measurement Index (-)",
    "y_label": "Temperature (°C)",
    "chart_type": "line+markers",
    "x": clean_time.tolist(),
    "y": clean_temp.tolist(),
}

# self-check
print("labels_ready:", all([plot_spec["title"], plot_spec["x_label"], plot_spec["y_label"]]))
print("same_length:", len(plot_spec["x"]) == len(plot_spec["y"]))

파이프라인 출력 예시

5단계 흐름을 따라가면, 센서 데이터 분석 결과가 어떤 그래프 구조로 표현되는지 확인할 수 있습니다.

센서 데이터 분석 파이프라인 결과

차트를 불러오는 중...

안내 예제 · 전체 파이프라인 미리보기

안내 예제 · 5단계 파이프라인 미리보기 코드를 실행하고 출력 결과를 확인하세요.

import numpy as np

# ===== Step 1: 데이터 생성 =====
rng = np.random.default_rng(42)
n = 24
raw_time = np.arange(n)
raw_temp = 22.0 + rng.normal(scale=1.5, size=n)

# 불량 데이터 주입: 정상 범위를 벗어나므로 제거 대상
raw_temp[3]  = -5.0   # 음수 오류
raw_temp[17] = 80.0   # 극단값 오류

# 유효하지만 낮거나 높은 값: 분류 결과를 보기 위한 정상 범위 데이터
raw_temp[8] = 17.5
raw_temp[12] = 27.2

print(f"원본 데이터 (n={n}):")
print("  처음 6개:", np.round(raw_temp[:6], 2))
print(f"  불량 포함: index 3={raw_temp[3]}, index 17={raw_temp[17]}")
print(f"  유효한 낮음/높음: index 8={raw_temp[8]}, index 12={raw_temp[12]}")

# ===== Step 2: 처리 함수 =====
def make_valid_mask(data, low=0.0, high=50.0):
  return (data >= low) & (data <= high)

def filter_valid(data, valid_mask):
  return data[valid_mask]

def compute_stats(data):
  return {
      "mean": round(float(data.mean()), 2),
      "std":  round(float(data.std()),  2),
      "min":  round(float(data.min()),  2),
      "max":  round(float(data.max()),  2),
  }

def classify_temp(temp_c):
  if temp_c < 18:
      return "cold"
  elif temp_c <= 26:
      return "normal"
  else:
      return "hot"

# ===== Step 3: 검증 =====
def checked_mean(data):
  if len(data) == 0:
      print("경고: 정상 범위 데이터가 없습니다")
      return None
  return float(data.mean())

valid_mask = make_valid_mask(raw_temp)
clean_temp = filter_valid(raw_temp, valid_mask)
clean_time = raw_time[valid_mask]

assert len(clean_temp) > 0
assert len(clean_time) == len(clean_temp)
print(f"\n정제 후: {len(raw_temp)}개 → {len(clean_temp)}개 (제거: {len(raw_temp)-len(clean_temp)}개)")
print("남은 측정 인덱스:", clean_time.tolist())

# ===== Step 4: 출력 =====
stats = compute_stats(clean_temp)
print("\n요약 통계:")
print(f"  {'항목':<6} {'값':>7}")
print("  " + "-" * 14)
for k, v in stats.items():
  print(f"  {k:<6} {v:>7.2f}")

# 각 측정값 분류
labels = [classify_temp(v) for v in clean_temp]
label_counts = {}
for lb in labels:
  label_counts[lb] = label_counts.get(lb, 0) + 1
print("\n분류 결과:", label_counts)

m = checked_mean(clean_temp)
print(f"\n해석: 정제된 {len(clean_temp)}개 측정값의 평균 온도는 {m:.2f}°C이다.")

# ===== Step 5: 시각화 설계안 =====
plot_spec = {
  "title": "Sensor Temperature over 24 Readings",
  "x_label": "Measurement Index (-)",
  "y_label": "Temperature (°C)",
  "chart_type": "line+markers",
  "x": clean_time.tolist(),
  "y": [round(float(v), 2) for v in clean_temp],
}
print("\n시각화 설계안 self-check:")
print("  labels_ready:", all([plot_spec["title"], plot_spec["x_label"], plot_spec["y_label"]]))
print("  same_length:", len(plot_spec["x"]) == len(plot_spec["y"]))
print("  point_count:", len(plot_spec["x"]))
print("\n전체 파이프라인 흐름 확인")

코드 흐름 따라 읽기 · 단계별 작성 워크스루

목표: 파이프라인 Step 1에서 센서 데이터가 만들어지는 흐름을 직접 작성합니다.

해야 할 일: 24개 온도 데이터를 만들고, 음수 오류와 극단값 오류를 각각 1개씩 주입한 뒤 기본 정보를 출력하세요.

완료 조건: shape, dtype, 처음 6개 값, 불량값 위치, assert 통과 문장이 출력되어야 합니다.

실습 1 · 센서 데이터 생성 코드를 실행하고 출력 결과를 확인하세요.

목표: 파이프라인 Step 2에서 처리 함수를 역할별로 나누어 구현합니다.

해야 할 일: 필터링, 통계 계산, 온도 분류 함수를 완성하고 정제 데이터에 차례대로 적용하세요.

완료 조건: 정제 전/후 개수, 통계 딕셔너리, 분류 샘플, assert 통과 문장이 출력되어야 합니다.

실습 2 · 처리 함수 구현 코드를 실행하고 출력 결과를 확인하세요.

import numpy as np

rng = np.random.default_rng(7)
raw_time = np.arange(24)
raw_temp = 21.5 + rng.normal(scale=2.0, size=24)
raw_temp[5] = -3.0
raw_temp[20] = 75.0

def make_valid_mask(data, low=0.0, high=50.0):
  # TODO 1) (data >= low) & (data <= high) 를 반환하세요.
  pass

def filter_valid(data, valid_mask):
  # TODO 2) data[valid_mask] 를 반환하세요.
  pass

def compute_stats(data):
  # TODO 3) mean, std, min, max 를 딕셔너리로 반환하세요.
  # 힌트) round(float(data.mean()), 2) 형태를 사용합니다.
  pass

def classify_temp(temp_c):
  # TODO 4) 18 미만 cold, 18~26 normal, 26 초과 hot 을 반환하세요.
  pass

# TODO 5) 아래 흐름이 실행되도록 위 함수들을 완성하세요.
valid_mask = make_valid_mask(raw_temp)
clean_temp = filter_valid(raw_temp, valid_mask)
clean_time = raw_time[valid_mask]
stats = compute_stats(clean_temp)
label_sample = [classify_temp(v) for v in clean_temp[:5]]

print(f"정제 전/후: {len(raw_temp)}개 → {len(clean_temp)}개")
print("통계:", stats)
print("분류 샘플:", label_sample)

assert len(clean_temp) == 22
assert len(clean_time) == len(clean_temp)
print("검증: 불량값 2개 제거와 시간 인덱스 길이가 맞습니다.")

목표: 파이프라인 Step 3에서 빈 데이터 검증 정책을 함수로 구현합니다.

해야 할 일: checked_mean이 정상 배열에서는 평균을 반환하고, 빈 배열에서는 None을 반환하게 하세요.

완료 조건: 정상 케이스 평균, 빈 배열 결과, assert 통과 문장이 출력되어야 합니다.

실습 3 · 빈 데이터 검증 코드를 실행하고 출력 결과를 확인하세요.

목표: 파이프라인 Step 4에서 통계와 분류 결과를 사람이 읽는 리포트로 바꿉니다.

해야 할 일: f-string 테이블을 출력하고, 온도 분류 결과를 카운팅한 뒤 가장 많은 분류를 해석하세요.

완료 조건: 통계 테이블, 분류별 개수, 가장 많은 분류, 해석 문장이 출력되어야 합니다.

실습 4 · f-string 리포트 출력 코드를 실행하고 출력 결과를 확인하세요.

import numpy as np

rng = np.random.default_rng(7)
raw_temp = 21.5 + rng.normal(scale=2.0, size=24)
raw_temp[5] = -3.0
raw_temp[20] = 75.0
valid_mask = (raw_temp >= 0.0) & (raw_temp <= 50.0)
clean_temp = raw_temp[valid_mask]
assert len(clean_temp) > 0
stats = {
  "mean": round(float(clean_temp.mean()), 2),
  "std": round(float(clean_temp.std()), 2),
  "min": round(float(clean_temp.min()), 2),
  "max": round(float(clean_temp.max()), 2),
}

def classify_temp(temp_c):
  if temp_c < 18:
      return "cold"
  elif temp_c <= 26:
      return "normal"
  else:
      return "hot"

# 힌트 1) f"{key:<8} {val:>8.2f}" 로 표를 출력하세요.
# 힌트 2) label_counts = {} 로 분류 결과를 카운팅하세요.
# 힌트 3) def count_key(label): return label_counts[label] 형태를 사용할 수 있습니다.

# TODO 1) stats 를 f-string 표로 출력하세요.
# TODO 2) labels = [classify_temp(v) for v in clean_temp] 를 만드세요.
# TODO 3) label_counts 딕셔너리로 분류별 개수를 세세요.
# TODO 4) most_common_label 을 찾고 해석 문장을 출력하세요.
# TODO 5) assert sum(label_counts.values()) == len(clean_temp) 로 검증하세요.

목표: Step 1~5를 하나의 파이프라인으로 연결해 전체 흐름을 완성합니다.

해야 할 일: 데이터 생성 → 처리 함수 → 검증 → 출력 → 시각화 설계안 순서로 코드를 채우고, 각 단계의 검증 결과를 출력하세요.

완료 조건: 통계 테이블, 분류 카운트, plot_spec self-check, 관찰/원인/시사점 3문장, assert 통과 문장이 출력되어야 합니다.

실습 5 · 전체 파이프라인 통합 코드를 실행하고 출력 결과를 확인하세요.

import numpy as np

# 1단계: 데이터 생성
rng = np.random.default_rng(7)
raw_time = np.arange(24)
raw_temp = 21.5 + rng.normal(scale=2.0, size=24)
raw_temp[5] = -3.0
raw_temp[20] = 75.0

# 2단계: 처리 함수
def make_valid_mask(data, low=0.0, high=50.0):
  # TODO 1) 정상 범위 mask 를 반환하세요.
  pass

def filter_valid(data, valid_mask):
  # TODO 2) mask 가 True 인 값만 반환하세요.
  pass

def compute_stats(data):
  # TODO 3) mean, std, min, max 딕셔너리를 반환하세요.
  pass

def classify_temp(temp_c):
  # TODO 4) cold / normal / hot 중 하나를 반환하세요.
  pass

# 3단계: 검증과 정제
assert raw_temp.shape == (24,)
valid_mask = make_valid_mask(raw_temp)
clean_temp = filter_valid(raw_temp, valid_mask)
clean_time = raw_time[valid_mask]

# TODO 5) clean_temp, clean_time 길이를 assert 로 검증하세요.

# 4단계: 출력 리포트
stats = compute_stats(clean_temp)
labels = [classify_temp(v) for v in clean_temp]
label_counts = {}

# TODO 6) stats 표를 출력하세요.
# TODO 7) label_counts 를 완성하고 출력하세요.

# 5단계: 시각화 설계안
plot_spec = {
  "title": "Sensor Temperature over 24 Readings",
  "x_label": "Measurement Index (-)",
  "y_label": "Temperature (°C)",
  "chart_type": "line+markers",
  "x": [],  # TODO 8-1) clean_time.tolist() 로 바꾸세요.
  "y": [],  # TODO 8-2) [round(float(v), 2) for v in clean_temp] 로 바꾸세요.
}

labels_ready = all([plot_spec["title"], plot_spec["x_label"], plot_spec["y_label"]])
same_length = len(plot_spec["x"]) == len(plot_spec["y"])
point_count = len(plot_spec["x"])

print("plot_spec self-check")
print("  labels_ready:", labels_ready)
print("  same_length:", same_length)
print("  point_count:", point_count)

# TODO 9) labels_ready, same_length, point_count == len(clean_temp) 를 assert 로 검증하세요.
# TODO 10) 관찰/원인/시사점 3문장을 출력하세요.

누적 복습 체크리스트

이 워크스루에서 한 번에 등장한 개념을 확인해 보세요.

주차	핵심 개념	이 워크스루에서 쓴 곳
W2	변수·단위	`base_temp = 22.0` (°C), `low = 0.0`, `high = 50.0`
W4	반복문	`for v in clean_temp:`
W5	함수	`filter_valid`, `compute_stats`, `classify_temp`
W6	자료구조	`stats` dict, `label_counts` dict
W9	리팩토링	함수 분해, `assert` 검증
W10	행렬·벡터 기초	데이터 표의 행/열, shape 해석
W11	NumPy	`np.array`, 불리언 mask, shape 검증
W12	시각화	`rng.normal` 노이즈 생성, `plot_spec` 딕셔너리, self-check
W13	알고리즘	`label_counts` 카운팅, `classify_temp` 분류

흐름을 따라간 뒤 확인할 출력

전체 예시 코드를 실행했을 때 아래 항목이 보이면 흐름을 제대로 따라간 것입니다.

정제 후: 24개 → 22개
요약 통계: mean/std/min/max 표
분류 카운트: {'cold': ..., 'normal': ..., 'hot': ...}
labels_ready: True
same_length: True
전체 파이프라인 흐름 확인

주차별 학습 전체 주차 목록으로 돌아가기

실습 콘솔 학생용 빈 Python 콘솔에서 자유롭게 실습하기

Debug Labs 앱에서 step-by-step으로 코드 흐름 보기