1. 프로젝트 개요
프로젝트 명:
•
시나리오 기반 제어 시스템 - SCC 파생 프로젝트
프로젝트 목적:
•
시나리오 기반의 제어 시스템
•
작은 코드 조각들을 조합하여 복잡한 시스템을 구현하는 구조에 대한 샘플 구조
주요 목표:
•
중앙 상태 저장소 기반의 시나리오 관리
◦
모든 코드 조각이 중앙 상태 저장소에 접근하여 상태를 확인하고, 변경하며, 제어하는 구조
•
개발 및 운영 환경의 통합
◦
개발자가 Theia IDE를 사용하여 즉각적인 테스트 및 운영 환경 반영 가능.
2. 시스템 개념 및 아키텍처
2.1 시나리오 기반 제어
코드 조각들이 이벤트 및 상태를 모니터링하고, 필요 시 중앙 저장소를 통해 동작 수행.
예시 시나리오 구성:
•
코드 조각 0 : 온도 수집
•
코드 조각 1: "온도 100도가 되면 난로를 꺼라"
•
코드 조각 2: "온도 0도가 되면 난로를 켜라"
확장 가능 예시:
•
특정 무게가 확인되고 사라지면 물품이 추가된 것으로 기록.
3. 아키텍처 설계
3.1 주요 구성 요소
구성 요소 | 설명 |
Theia IDE | 개발자가 웹에서 코드 작성 및 테스트 가능 |
중앙 상태 저장소 | 중앙 데이터 기반 상태 관리 |
시나리오 확인 및 조회, 제어 | Python 기반 코드 조각을 통한 작업 처리
- 데이터 확인
- 데이터 가공
- 제어 명령 |
3.2 시스템 흐름
1.
시나리오 작성 및 테스트
•
개발자는 Theia IDE에서 시나리오 작성.
2.
시나리오 실행 및 상태 모니터링
•
시나리오 실행기가 이벤트 감지 및 실행.
•
중앙 상태 저장소에 데이터 기록 및 변경.
3.
실시간 대시보드 모니터링
•
웹 대시보드를 통해 상태 모니터링 및 제어.
•
REST API를 통한 외부 서비스 연동.
4. 시나리오 실행 예시
코드 조각 예시: 온도 제어 시나리오
from prefect import task, flow
import time
sample_dataspace = {
"stove": False,
"temperature": 0
}
# temp simulation
@task
def sample_simulation_stove():
global sample_dataspace
if sample_dataspace["stove"]:
sample_dataspace["temperature"] += 1
else:
sample_dataspace["temperature"] -= 1
print("temperature")
print(sample_dataspace["temperature"])
# check
@task
def check_temperature():
global sample_dataspace
return sample_dataspace["temperature"]
# control
@task
def control_heater(temp):
global sample_dataspace
if temp > 100:
print("stove off")
sample_dataspace["stove"] = False
elif temp < 0:
print("stove on")
sample_dataspace["stove"] = True
@flow
def temperature_control_flow():
while True:
temp = check_temperature()
control_heater(temp)
# temp code
sample_simulation_stove()
time.sleep(1)
if __name__ == "__main__":
temperature_control_flow()
Python
복사
실행 및 배포
python3 temperature_control.py
prefect deploy --name temp-control --pool scc-process-pool
Shell
복사
5. 기대 효과
1.
개발 및 운영의 효율성 증가
•
개발자가 직접 코드를 기반으로 운영 가능.
•
빠른 기능 추가 및 변경 용이.
2.
시스템의 확장성 및 유지보수 용이성 확보
•
모듈화된 코드 조각으로 기능 확장 가능.
•
상태 저장소 기반의 일관된 관리 가능.
3.
실시간 대응 및 자동화된 관리 체계 구축
•
시나리오 기반의 자동화된 제어 및 실행.
•
장애 발생 시 즉각적인 복구 및 대응.
6. 시나리오 실행 예시
코드 조각 예시: 개수 측정 시나리오
from prefect import task, flow
import time
import random
sample_dataspace = {
"status": False,
"count": 0
}
# check
@task
def check_weight():
return random.choice([0, 100])
@flow
def count_flow():
global sample_dataspace
while True:
weight = check_weight()
if weight >= 100:
sample_dataspace["status"] = True
elif weight <= 0:
if sample_dataspace["status"]:
sample_dataspace["count"] += 1
sample_dataspace["status"] = False
time.sleep(1)
if __name__ == "__main__":
count_flow()
Python
복사
