센서 로그 저장과 조회
학습 목표
같은 Key로 다시 저장할 때 새 기록이 생기는지 기존 기록이 바뀌는지, 장치 로그의 구조로 확인합니다.
실행 조건
Python의 sqlite3를 사용하며 DB 서버·계정은 필요하지 않습니다.
중복 측정은 한 번만 저장하기
시작 조건
장치 m1의 s 측정값은 20,21,22,23입니다. 같은 장치·시각은 최신 값으로 갱신합니다.
진행 방법
- SQLite의 복합 Primary Key를 확인하고 제공된 네 측정값을 저장하세요.
- 저장된 행 수와 의 값을 확인하고, 중복 처리 규칙이 맞는지 설명하세요.
남길 결과
- 저장 결과와 중복 처리 예 1개.
import sqlite3
connection=sqlite3.connect(':memory:')
connection.execute('CREATE TABLE readings(device TEXT,t REAL,value REAL,PRIMARY KEY(device,t))')
records=[('m1',0,20.),('m1',1,21.),('m1',1,22.),('m1',2,23.)]
connection.executemany('INSERT INTO readings VALUES(?,?,?) ON CONFLICT(device,t) DO UPDATE SET value=excluded.value',records)
connection.commit()
print(connection.execute('SELECT * FROM readings ORDER BY t').fetchall())
결과 기록·자가 점검
입력 수:
저장 행 수:
중복 Key:
최종 값:
확인할 점
- 측정 시각과 수신 시각을 혼동하지 않았는가?
- SQL 문자열에 값을 직접 붙이지 않고 Parameter를 사용했는가?
시간 구간과 장치로 조회하기
시작 조건
m2 장치에 , 값 30,31,32를 추가합니다. m1의 1≤t≤2만 조회합니다.
진행 방법
- 장치·시작 시각·종료 시각을 인자로 받는 조회 함수를 작성하세요.
- 지정 구간과 값이 없는 구간을 한 번씩 시험하세요.
- 조회 결과를 시간–측정값 그래프로 그리세요.
남길 결과
- 조회 함수와 두 조회 결과, 시계열 그림.
connection.executemany('INSERT INTO readings VALUES(?,?,?)',[('m2',0,30.),('m2',1,31.),('m2',2,32.)])
def query(device,start,end):
return connection.execute('SELECT t,value FROM readings WHERE device=? AND t BETWEEN ? AND ? ORDER BY t',(device,start,end)).fetchall()
print(query('m1',1,2));print(query('m1',10,20))
# TODO: 시계열을 그리세요.
결과 기록·자가 점검
장치:
조회 구간:
행 수:
값의 범위:
확인할 점
- 정렬 순서를 SQL에서 명시했는가?
- 빈 결과를 오류나 값 0으로 바꾸지 않았는가?
센서에서 API까지 연결하기
시작 조건
다음 단원의 HTTP API가 조회 함수를 호출한다고 가정합니다. 요청은 device=m1,start=0,end=2입니다.
진행 방법
- 조회 결과를 JSON으로 전달할 필드 구조를 정하세요.
- 없는 장치와 start>end를 구분해 응답 규칙을 정하세요.
- DynamoDB 등 외부 저장소로 바꾸어도 유지할 데이터·조회 규칙을 표시하세요.
남길 결과
- 두 정상·오류 요청의 응답 예와 저장소 교체 시 유지할 규칙.
import json
print(json.dumps({'device':'m1','readings':[{'time_s':t,'value':v} for t,v in query('m1',0,2)]}))
# TODO: 입력 검증과 오류 응답을 작성하세요.
결과 기록·자가 점검
요청:
검증 결과:
응답 필드:
예상 행 수:
확인할 점
- 외부 서비스 계정 없이도 저장·조회 원리를 확인했는가?
- 장치 ID와 측정 시각의 의미를 API에서도 유지했는가?