3. 블루투스 시리얼 통신

• 라즈베리파이의 시리얼 통신기능을 이용하여 블루투스 시리얼 통신으로 스마트폰과 데이터를 주고 받는다.

• 블루투스 시리얼 통신 모듈은 라즈베리파이의 시리얼 통신으로 손쉽게 블루투스 통신으로 데이터를 주고 받을 수 있다.

라즈베리파이에서 시리얼 통신을 사용하기 위해서는 OS상에서 통신 기능을 활성화시켜야 한다.

1. VNC를 사용한 확인
   라즈베리 아이콘 -> Preference -> Raspberry Pi Configuration
   
   [Interfaces]
   

• Serial Port : Enable
• Serial Console : Disable ( Serial Port를 Enable 시에 활성화됨 )
  

이후 재부팅

1. putty 터미널을 사용한 확인

• sudo raspi-config ->
• 1. Interfaces Options ->
• I6. Serial Port ->
• 첫 번째 나오는 창은 <아니오> ->
• 두 번째 나오는 창은 <예> ->
• Finish
• reboot 권고 명령에서 <예>

3-1. 시리얼 포트 확인

• 라즈베리파이는 GPIO의 15번과 14번을 시리얼 통신용으로 할당되어 있다.
• 리눅스 터미널에서 ls -al /dev/serial0 명령어로 확인해 볼 수 있다.

• 위 처럼 현재 serial0이 ttyS0으로 연결되어 있다.

3-1-1. 스마트폰과의 통신을 위한 어플리케이션 설치

1. 안드로이드

채팅을 입력하는 것 처럼 입력 문자열을 라즈베리파이로 보낼 수 있다
또한 m1 ~ m7 단축 버튼을 사용할 수 있다

2. 아이폰

3-2. 시리얼 통신 (데이터 받기)

AttributeError: partially initialized module 'serial' has no attribute 'Serial' (most likely due to a circular import)

• 위와 같은 에러 코드가 발생 할 시의 대처법
  • 오타 확인
  • pip uninstall pyserial
  • pip uninstall serial
  • 이후 serial.py 재실행
  • (같은 오류로 재실행 실패 시에)
    • pip install pyserial --force
    • pip install serial --force

한 글자씩 받기 (ser.py)

# 한 글자씩 받기
import serial

se = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
# baudrate = 통신 속도
# timeout = 1.0초 동안 데이터가 입려되지 않으면 timeout 이벤트 발생

try:
    while True:
        data = se.read()
        print(data)
        
except KeyboardInterrupt:
    pass

se.close()

한 line 씩 받기 (ser2.py)

import serial

se = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)


try:
    while True:
        data = se.readline()
        data = data.decode() # \r\n 과 같은 정규표현식을 decode 하여 출력한다.
        print(data)
        
except KeyboardInterrupt:
    pass

se.close()

해당 시리얼 통신으로 스마트폰 - 라즈베리파이가 연결되었으며 이를 putty를 통해 개발자의 pc 모니터로 확인할 수 있다.

3-3. 시리얼 통신 (데이터 보내기)

시리얼 통신 코드의 작성 (ser3.py)

• 아이폰의 application 에서는 raspberry 에서 보내진 데이터를 확인하는 기능을 제공하지 않는다.

import serial
import time

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)

try:
    while True:
        data = "Hello world \r\n"
        ser.write(data.encode())
        time.sleep(1.0)
        
except KeyboardInterrupt:
    pass

ser.close()

3-4. 시리얼 통신 (응용)

• 시리얼 통신을 통해 무선 기기에서 raspberry pi로 송출한 데이터에서 필요한 데이터를 search해 다양한 기능을 사용해보자.

통신 받은 data에서 문자열 확인하기 (ser4.py)

import serial

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)

try:
    while True:
        data = bleSerial.readline()
        data = data.decode()
        
        if data.find("go") >= 0: # find() 함수는 문자열 중 원하는 문자열을 찾으면 0, 그렇지 못했다면 -1을 반환한다.
            print("gogo")
        elif data.find("back") >= 0:
            print("back")
        elif data.find("left") >= 0:
            print("left")
        elif data.find("right") >= 0:
            print("right")
        elif data.find("stop") >= 0:
            print("stop")
            
except KeyboardInterrupt:
    pass
ser.close()

3-5. 쓰레드 활용하기

• 쓰레드를 활용하여 좀 더 자유로운 코드를 작성하자

쓰레드 : 프로그램을 실행하는 일꾼을 추가하여 둘 이상의 코드를 동시에 실행시키는 기능

달리기로 예를 들어보자

• 이어달리기 : 쓰레드를 사용하지 않은 프로그램
  • 한 명의 주자가 도착해야 다음 주자가 출발 할 수 있다.
  • 하나의 코드 동작이 완료된 후에 다음 코드를 실행한다.

• 달리기 경주 : 쓰레드를 사용한 프로그램
  • 여러 명의 주자가 동시에 (혹은 시간차를 두고) 출발 하여 각자 도착점까지 달려간다.
  • 하나의 코드를 수행하는 쓰레드와 또 다른 하나의 코드를 수행하는 쓰레드들이 각자 코드를 실행한다.

import threading
import time

def hello():
    for i in range(10):
        print("hello")
        time.sleep(0.2)
        
def main():
    for i in range(10):
        print("main code")
        time.sleep(0.2)

# 스레드를 사용하지 않았을 시에
hello()
main()

# 스레드를 사용할 시에
task1 = threading.Thread(target=hello)
task1.start()
main()

쓰레드가 필요 할 만한 코드 (thr1.py)

import serial
import time

se = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)


try:
    while True:
        data = se.readline()
        data = data.decode() # \r\n 과 같은 정규표현식을 decode 하여 출력한다.
        print(data)
        
        for i in range(10):
            print(i)
            time.sleep(0.15)
        
except KeyboardInterrupt:
    pass

se.close()

• 위의 코드를 실행하고 시리얼 통신을 통해 데이터를 입력받으려 하면 반복문 때문에 데이터가 밀려나가는 것을 볼 수 있다.

쓰레드를 사용한 코드 (thr2.py)

import serial
import time
import threading

se = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)

def serial():
    while True:
        data = se.readline()
        data = data.decode()
        print("serial : ", data)
    
def main():
    try:
        while True:
            print("hello")
            time.sleep(1.0)
            
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    se.close()

작업관리자 - 프로세스 목록 : 단독으로 실행되는 하나의 일

- 멀티 프로세스 : 홍길동 분신 같은거 
- 하나의 CPU에 다양한 프로세스 사용하기

멀티 쓰레드 : 하나의 프로세스를 나누어서 데이터 영역을 공유하면서 프로세스를 운영함.

프로세스 큰 개념 -> 쓰레드 작은 개념.

• 쓰레드를 사용한 프로그램을 ctrl + c로 종료하려고 하면 에러가 난다.
• main 프로그램은 main() 함수의 try except 문에 걸려 정상적으로 종료되지만 실행 중인 thread는 종료되지 않아 발생하는 에러이다.
• 하지만 신경쓰지 않아도 문제 없는 오류이다.

3-6. 블루투스 시리얼 통신을 이용한 무선 조종 구현하기

1단계. 틀 만들기 (car1.py)

import threading
import serial
import time

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
ser_data = "" # 시리얼 통신으로 데이터를 입력 받을 변수

def serial():
    global ser_data
    while True:
        data = ser.readline()
        data = data.decode()
        ser_data = data
        
def main():
    global ser_data
    try:
        while True:
            if ser_data.find("go") >= 0:
                print(ser_data)
                ser_data = ""
                
            elif ser_data.find("back") >= 0:
                print(ser_data)
                ser_data = ""
                
            elif ser_data.find("left") >= 0:
                print(ser_data)
                ser_data = ""
                
            elif ser_data.find("right") >= 0:
                print(ser_data)
                ser_data = ""
                
            elif ser_data.find("stop") >= 0:
                print(ser_data)
                ser_data = ""
                      
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    ser.close()

2단계. 기능 구현 (car2.py)

import RPi.GPIO as gpio
import threading
import serial
import time

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
ser_data = "" # 시리얼 통신으로 데이터를 입력 받을 변수

L_M = [18, 22, 27] #PWMA, AIN1, AIN2
R_M = [23, 25, 24] #PWMB, BIN1, BIN2

gpio.setmode(gpio.BCM)

for i in L_M:
    gpio.setup(i, gpio.OUT)
    
for i in R_M:
    gpio.setup(i, gpio.OUT)
    
L_Motor = gpio.PWM(L_M[0], 500)
L_Motor.start(0)

R_Motor = gpio.PWM(R_M[0], 500)
R_Motor.start(0)


def serial():
    global ser_data
    while True:
        data = ser.readline()
        data = data.decode()
        ser_data = data
        
def drive(speed, left, right):
    # 좌측 모터 드라이버
    gpio.output(L_M[1], left)
    
    if left <= 0 :
        left2 = 1
    else:
        left2 = 0
        
    gpio.output(L_M[2], left2)
    L_Motor.ChangeDutyCycle(speed)
    
    # 우측 모터 드라이버
    gpio.output(R_M[1], right)
    
    if right <= 0:
        right2 = 1
    else:
        right2 = 0
        
    gpio.output(R_M[2], right2)
    R_Motor.ChangeDutyCycle(speed)
    
        
def main():
    global ser_data
    try:
        while True:
            if ser_data.find("go") >= 0:
                print(ser_data)
                drive(50, 0, 0)
                ser_data = ""
                
            elif ser_data.find("back") >= 0:
                print(ser_data)
                drive(50, 1, 1)
                ser_data = ""
                
            elif ser_data.find("left") >= 0:
                print(ser_data)
                drive(50, 1, 0)
                ser_data = ""
                
            elif ser_data.find("right") >= 0:
                print(ser_data)
                drive(50, 0, 1)
                ser_data = ""
                
            elif ser_data.find("stop") >= 0:
                print(ser_data)
                drive(0, 0, 0)
                ser_data = ""
                      
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    gpio.cleanup()
    ser.close()

스위치를 이용한 비상 정지 기능 (car3.py)

• 블루투스 모듈과의 거리가 떨어지면 무선통신이 끊어질 수 있다.
• 무선통신은 항상 끊어짐에 대비해야하며 이를 대비하여 아날로그 스위치로 긴급 정지 기능을 구현하는 경우가 많다.

import RPi.GPIO as gpio
import threading
import serial
import time

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
ser_data = "" # 시리얼 통신으로 데이터를 입력 받을 변수

SW = [5,6,13,19] # sw1 , 2,  3, 4
L_M = [18, 22, 27] #PWMA, AIN1, AIN2
R_M = [23, 25, 24] #PWMB, BIN1, BIN2

gpio.setmode(gpio.BCM)

for i in SW:
    gpio.setup(i, gpio.IN, pull_up_down=gpio.PUD_DOWN)

for i in L_M:
    gpio.setup(i, gpio.OUT)
    
for i in R_M:
    gpio.setup(i, gpio.OUT)
    
L_Motor = gpio.PWM(L_M[0], 500)
L_Motor.start(0)

R_Motor = gpio.PWM(R_M[0], 500)
R_Motor.start(0)


def serial():
    global ser_data
    while True:
        data = ser.readline()
        data = data.decode()
        ser_data = data
        
def drive(speed, left, right):
    # 좌측 모터 드라이버
    gpio.output(L_M[1], left)
    
    if left <= 0 :
        left2 = 1
    else:
        left2 = 0
        
    gpio.output(L_M[2], left2)
    L_Motor.ChangeDutyCycle(speed)
    
    # 우측 모터 드라이버
    gpio.output(R_M[1], right)
    
    if right <= 0:
        right2 = 1
    else:
        right2 = 0
        
    gpio.output(R_M[2], right2)
    R_Motor.ChangeDutyCycle(speed)
    
        
def main():
    global ser_data
    try:
        while True:
            if ser_data.find("go") >= 0:
                print(ser_data)
                drive(50, 0, 0)
                ser_data = ""
                
            elif ser_data.find("back") >= 0:
                print(ser_data)
                drive(50, 1, 1)
                ser_data = ""
                
            elif ser_data.find("left") >= 0:
                print(ser_data)
                drive(50, 1, 0)
                ser_data = ""
                
            elif ser_data.find("right") >= 0:
                print(ser_data)
                drive(50, 0, 1)
                ser_data = ""
                
            elif ser_data.find("stop") >= 0:
                print(ser_data)
                drive(0, 0, 0)
                ser_data = ""
                
            # gpio.HIGH == 0
            if gpio.input(SW[0]) == gpio.HIGH or gpio.input(SW[1]) == gpio.HIGH or gpio.input(SW[2]) == gpio.HIGH or gpio.input(SW[3]) == gpio.HIGH:
                drive(0,0,0)

                      
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    gpio.cleanup()
    ser.close()

3-7. RC카에 다양한 기능 추가하기

이동방향 LED로 표시하기 (carAdd1.py)

• LED를 이동방향에 맞게 켜는 코드를 만들어보자

import threading
import serial
import time
import RPi.GPIO as gpio

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
ser_data = "" # 시리얼 통신으로 데이터를 입력 받을 변수

SW = [5,6,13,19] # sw1 , 2,  3, 4
L_M = [18, 22, 27] #PWMA, AIN1, AIN2
R_M = [23, 25, 24] #PWMB, BIN1, BIN2
LED = [26, 16, 20, 21] #전좌, 전우, 후좌, 후우


gpio.setmode(gpio.BCM)

for i in LED:
    gpio.setup(i, gpio.OUT)
#     gpio.setup(i, gpio.OUT, initial=gpio.LOW) # 프로그램 시작과 동시에 gpio 상태를 셋팅함, LOW 해두면 프로그램을 시작하면서 LED가 꺼진상태로 시작.

for i in SW:
    gpio.setup(i, gpio.IN, pull_up_down=gpio.PUD_DOWN)

for i in L_M:
    gpio.setup(i, gpio.OUT)
    
for i in R_M:
    gpio.setup(i, gpio.OUT)
    
L_Motor = gpio.PWM(L_M[0], 500)
L_Motor.start(0)

R_Motor = gpio.PWM(R_M[0], 500)
R_Motor.start(0)


def serial():
    global ser_data
    while True:
        data = ser.readline()
        data = data.decode()
        ser_data = data
        
def drive(speed, left, right):
    # 좌측 모터 드라이버
    gpio.output(L_M[1], left)
    
    if left <= 0 :
        left2 = 1
    else:
        left2 = 0
        
    gpio.output(L_M[2], left2)
    L_Motor.ChangeDutyCycle(speed)
    
    # 우측 모터 드라이버
    gpio.output(R_M[1], right)
    
    if right <= 0:
        right2 = 1
    else:
        right2 = 0
        
    gpio.output(R_M[2], right2)
    R_Motor.ChangeDutyCycle(speed)
    
    
def led_func(front_L, front_R, back_L, back_R):
    gpio.output(LED[0], front_L)
    gpio.output(LED[1], front_R)
    gpio.output(LED[2], back_L)
    gpio.output(LED[3], back_R)
    
        
def main():
    global ser_data
    try:
        while True:
            if ser_data.find("go") >= 0:
                print(ser_data)
                led_func(1,1,0,0)
                drive(50, 0, 0)
                ser_data = ""
                
            elif ser_data.find("back") >= 0:
                print(ser_data)
                led_func(0,0,1,1)
                drive(50, 1, 1)
                ser_data = ""
                
            elif ser_data.find("left") >= 0:
                print(ser_data)
                led_func(1,0,1,0)
                drive(50, 1, 0)
                ser_data = ""
                
            elif ser_data.find("right") >= 0:
                print(ser_data)
                led_func(0,1,0,1)
                drive(50, 0, 1)
                ser_data = ""
                
            elif ser_data.find("stop") >= 0:
                print(ser_data)
                led_func(0,0,0,0)
                drive(0, 0, 0)
                ser_data = ""
                
            # gpio.HIGH == 0
            if gpio.input(SW[0]) == gpio.HIGH or gpio.input(SW[1]) == gpio.HIGH or gpio.input(SW[2]) == gpio.HIGH or gpio.input(SW[3]) == gpio.HIGH:
                led_func(0,0,0,0)
                drive(0,0,0)
                

                      
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    gpio.cleanup()
    ser.close()

경적기능 추가 (carAdd2.py)

import threading
import serial
import time
import RPi.GPIO as gpio

import RPi.GPIO as gpio
import threading
import serial
import time

ser = serial.Serial("/dev/ttyS0", baudrate=9600, timeout=1.0)
ser_data = "" # 시리얼 통신으로 데이터를 입력 받을 변수

SW = [5,6,13,19] # sw1 , 2,  3, 4
L_M = [18, 22, 27] #PWMA, AIN1, AIN2
R_M = [23, 25, 24] #PWMB, BIN1, BIN2
LED = [26, 16, 20, 21] #전좌, 전우, 후좌, 후우
BUZZER = 12

gpio.setmode(gpio.BCM)

for i in LED:
    gpio.setup(i, gpio.OUT)
#     gpio.setup(i, gpio.OUT, initial=gpio.LOW) # 프로그램 시작과 동시에 gpio 상태를 셋팅함, LOW 해두면 프로그램을 시작하면서 LED가 꺼진상태로 시작.

for i in SW:
    gpio.setup(i, gpio.IN, pull_up_down=gpio.PUD_DOWN)

for i in L_M:
    gpio.setup(i, gpio.OUT)
    
for i in R_M:
    gpio.setup(i, gpio.OUT)
    
gpio.setup(BUZZER, gpio.OUT)
p = gpio.PWM(BUZZER, 391)
p.stop()

L_Motor = gpio.PWM(L_M[0], 500)
L_Motor.start(0)

R_Motor = gpio.PWM(R_M[0], 500)
R_Motor.start(0)


def serial():
    global ser_data
    while True:
        data = ser.readline()
        data = data.decode()
        ser_data = data
        
def drive(speed, left, right):
    # 좌측 모터 드라이버
    gpio.output(L_M[1], left)
    
    if left <= 0 :
        left2 = 1
    else:
        left2 = 0
        
    gpio.output(L_M[2], left2)
    L_Motor.ChangeDutyCycle(speed)
    
    # 우측 모터 드라이버
    gpio.output(R_M[1], right)
    
    if right <= 0:
        right2 = 1
    else:
        right2 = 0
        
    gpio.output(R_M[2], right2)
    R_Motor.ChangeDutyCycle(speed)
    
    
def led_func(front_L, front_R, back_L, back_R):
    gpio.output(LED[0], front_L)
    gpio.output(LED[1], front_R)
    gpio.output(LED[2], back_L)
    gpio.output(LED[3], back_R)
    
        
def main():
    global ser_data
    try:
        while True:
            if ser_data.find("go") >= 0:
                print(ser_data)
                led_func(1,1,0,0)
                drive(50, 0, 0)
                ser_data = ""
                
            elif ser_data.find("back") >= 0:
                print(ser_data)
                led_func(0,0,1,1)
                drive(50, 1, 1)
                ser_data = ""
                
            elif ser_data.find("left") >= 0:
                print(ser_data)
                led_func(1,0,1,0)
                drive(50, 1, 0)
                ser_data = ""
                
            elif ser_data.find("right") >= 0:
                print(ser_data)
                led_func(0,1,0,1)
                drive(50, 0, 1)
                ser_data = ""
                
            elif ser_data.find("stop") >= 0:
                print(ser_data)
                led_func(0,0,0,0)
                drive(0, 0, 0)
                ser_data = ""
                
            elif ser_data.find("bz_on") >= 0:
                print("bz_on")
                p.start(50)
                p.ChangeFrequency(391)
                ser_data = ""
            
            elif ser_data.find("bz_off") >= 0:
                print("bz_off")
                p.stop()
                ser_data = ""
            
                
                
            # gpio.HIGH == 0
            if gpio.input(SW[0]) == gpio.HIGH or gpio.input(SW[1]) == gpio.HIGH or gpio.input(SW[2]) == gpio.HIGH or gpio.input(SW[3]) == gpio.HIGH:
                led_func(0,0,0,0)
                drive(0,0,0)
                p.stop()
                

                      
    except KeyboardInterrupt:
        pass
    
if __name__ == '__main__':
    task1 = threading.Thread(target=serial)
    task1.start()
    main()
    gpio.cleanup()
    ser.close()

3-8. 라즈베리파이 부팅 시, 자동 코드 실행

putty 터미널

1. sudo nano /etc/xdg/lxsession/LXDE-pi/autostart

2. open된 nano 에디터의 이미 작성되어 있는 텍스트의 마지막 줄에 lxterminal -e sudo python3 /home/pi/경로/파일명.py 입력 - 해당 코드 입력 후 ctrl+o -> enter 로 저장
- 저장 후에 ctrl+x 나가기
- 저장하고 싶지 않다면 ctrl+o 입력 전에 ctrl+x로 나가기. (이 때 한 차례 저장을 할 것인지 물어보는데 이 때 n을 입력)

3. 재부팅 sudo reboot

4. 재부팅 후 VNC로 원격 GUI를 확인해 보면 별다른 조작 없이도 프로그램이 실행되어 있는 것을 확인 할 수 있다