Device Program

1. Device 화면에 이미지를 출력한다.
2. 소켓 통신을 통해 "Subscribe" 컴포넌트로 부터 메시지를 받는다.
   • 이 때, subscribe 컴포넌트가 client --> device가 server
3. 소켓 통신을 통해 "Publish" 컴포넌트로 메시지 보낸다.
   • 이 때, publish 컴포넌트가 server <-- device가 client

• 위 세가지 역할은 multi-processing을 통해 처리한다.

iot_device.py

from multiprocessing import Process, Semaphore, Array
import socket
import json
from time import sleep
import cv2

shared_arr = Array('i', range(10))	# Process간 데이터를 공유하기 위한 array 생성
# Array('i', range(10))은 int형 변수 10개 크기의 array를 생성하는 생성자
# 현재는 int형 변수 하나만 있어도 되지만, 프로그램의 요구 사항에 따라 다른 IPC 방식을 사용 할 수 있다.

sem = Semaphore()
# 공유 데이터의 동기화를 위한 semaphore

# 서로 다른 역할을 하는 프로세스 생성
Print_image = Process(target=device_control, args=(shared_arr, sem))
Action_trigger = Process(target=device_server, args=(shared_arr, sem))
Report_status = Process(target=device_client, args=(shared_arr, sem))

Print_image.start()
Action_trigger.start()
Report_status.start()

Print_image.join()
Action_trigger.join()
Report_status.join()

# ============================ openCV process ============================ #
def device_control(arr, sem):
    cv2.namedWindow("output", cv2.WINDOW_NORMAL)
    cv2.resizeWindow("output", 1200, 1150)
    while True:
        sem.acquire()
        ii = arr[0] # Read current image index from shared memory
        sem.release()
        imgFile = "Pictures/image" + str(ii) +".jpg"
        img = cv2.imread(imgFile, cv2.IMREAD_COLOR)
        cv2.imshow("output", img)
        ret = cv2.waitKey(1000)
    cv2.destroyALLWindows()
    
    
# ========================= Action_trigger process ======================== # 
def device_server(arr, sem):
    S_HOST = '192.168.0.20' # 디바이스(라즈베리파이) 내부 IP
    S_PORT = 8888           # 디바이스(라즈베리파이) 내부 PORT
 
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((S_HOST, S_PORT))
    server_socket.listen()
 
    while True:
        client_socket, addr = server_socket.accept() 
        # Greengrass Core Device(Client)가 연결 될 때까지 기다린다.
        
        print('connected by', addr)
        while True:
            data = client_socket.recv(1024)
            # data를 받는다.
            
            print(data.decode())
            if not data:
                print('Disconnect')
                break
            client_socket.sendall(data)
            # client에게 data 전체를 전송한다.
            
            desired = json.loads(data.decode()) # json 파싱
            sem.acquire()
            arr[0] = int(desired["content"]) # json에서 "content"의 내용 저장
            print(f'Get action as {arr[0]}')
            sem.release()
        client_socket.close()
    server_socket.close()    
    
# ======================== device_clinet =================================== #
def device_client(arr, sem):
    HOST = '10.177.xxx.xxx' # core device's IP
    PORT = 8888             # core device's PORT
 
    RPI_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    while True:
        try:
            RPI_client_socket.connect((HOST, PORT)) 
            # Greengrass Core Device로 메세지 보내기 위한 client 소켓 생성
            break
        except Exception as ex:
            print(f'Failed connect to server, retry after 5 sec {ex}')
            sleep(5)
 
    message = {"device": 1, "status": 0, "db_action": "put"}
    while True:       
        sem.acquire()
        message["status"] = arr[0]
        sem.release()
        RPI_client_socket.sendall(json.dumps(message).encode('utf-8'))
        # Core Device로 JSON 데이터를 송신한다.
        reply = RPI_client_socket.recv(1024)
        sleep(10)	# 10초마다 JSON 데이터 수정, 송신 반복
    RPI_client_socket.close()