본문 바로가기
  • SDXL 1.0 + 한복 LoRA
  • SDXL 1.0 + 한복 LoRA
Study/라즈베리파이

Firmata 공부하기 #4 - Firmata 코드 살펴보기 (1)

by 마즈다 2021. 2. 15.
반응형

이제부터는 본격적으로 Firmata를 분석해보고 사용해보도록 하자.

나의 개인적인 취향 상 분석을 먼저 면밀하게 한 후 사용을 하는 것 보다는 실제 동작하는 모습을

보면서 연관된 코드를 뜯어보는 것이 수월하기에 이번 포스팅 역시 그러한 방식으로 진행을 하겠다.

 

Firmata의 전체적인 구성은 호스트 PC 쪽의 Firmata Client와 아두이노쪽의 Firmata 스케치로 구성되어

있다. 물론 아두이노쪽에서 범용 Firmata 스케치가 아닌 Firmata 라이브러리를 사용한다면 구성이 조금

달라지겠지만 이 시리즈에서는 범용 Firmata 스케치를 사용한다는 전제로 포스팅을 하고 있다.

 

이러한 전제 하에 지난 번 포스팅의 가장 마지막에 실행했던 digital_output.py 소스를 가지고 간단하게

Firmata의 흐름과 관련 소스를 살펴보도록 하겠다.

 

다만 모든 소스를 세세하게 분석하는데는 많은 시간과 지식이 필요하므로 이번 포스팅에서는 우선 프로토콜이

어떻게 구현되는 지에 초점을 맞춰 살펴보도록 하자.

 

Firmata Client - pymata4

 

이전 포스팅에서도 언급한 바와 같이 Firmata Client는 다양한 언어로 구현이 가능하다. 그리고 이 시리즈

에서는 Python으로 된 Client를 사용하고 있으며 그 중에서도 자료가 가장 잘 정리되어있는 pymata4를

대상으로 한다.

 

pymata4의 핵심 소스는 3개의 파일로 구성이 되어있다.

 

  • pin_data.py : 아두이노의 pin mode를 설정하기 위한 각종 속성이 정의되어 있다.

  • private_constants.py : Firmata 프로토콜을 정의하는 여러가지 상수가 정의되어 있다.

  • pymata4.py : 가장 핵심적인 소스로 사용자들인 호출하여 사용할 API들이 구현되어있다.

 

단순히 Firmata를 이용하기 위해서는 pymata4.py에 있는 API를 어떻게 호출하는 지만 알면 된다. 이제

예제 코드인 digital_output.py로부터 시작하여 차근차근 알아보자.

 

digital_output.py 코드는 매우 단순하다 (코드에 집중하기 위해 원래 있던 주석은 모두 삭제하였다).

사전 이해를 위해 간단하게 코드에 주석을 붙였다. 

import sys
import time

from pymata4 import pymata4

# 사용할 디지털 핀 번호 설정. 기본 예제 코드에서는 6번을 사용한다.
DIGITAL_PIN = 6  # arduino pin number

# blink라는 함수를 구현하였다. 실제로 Firmata를 구동시키는 함수이다.
def blink(my_board, pin):
    # 가장 먼저 PIN 모드를 설정한다. 파라미터로 전달받은 핀(6번 핀)을 출력 모드로 설정한다. 
    my_board.set_pin_mode_digital_output(pin)

    # 1초 간격으로 6번 핀에 HIGH(1), LOW(0) 신호를 4회 반복하여 보낸다.
    # 이 6번 핀에 LED를 연결하여 1초 간격으로 LED가 켜지고 꺼지는 것을 4회 반복한다.
    # 직전 포스팅의 마지막 영상 참조.
    for x in range(4):
        print('ON')
	# 아두이노 보드의 6번 핀에 HIGH(1) 신호를 보낸다. LED가 연결된 경우 LED가 켜진다.
        my_board.digital_write(pin, 1)
        time.sleep(1)
        print('OFF')
	# 아두이노 보드의 6번 핀에 LOW(0) 신호를 보낸다. LED가 연결된 경우 LED가 꺼진다.
        my_board.digital_write(pin, 0)
        time.sleep(1)
    # 아두이노와의 연결을 끊는다.
    my_board.shutdown()

# 아두이노 보드의 인스턴스를 생성한다.
board = pymata4.Pymata4()
try:
    # blink 함수를 호출한다. 위에서 생성된 아두이노 보드 인스턴스와 사용할 핀 번호를 
    # 파라미터로 전달한다.
    blink(board, DIGITAL_PIN)
except KeyboardInterrupt:
    board.shutdown()
    sys.exit(0)

 

이제 위의 코드를 차근차근 살펴보자.


보드(아두이노) 인스턴스 생성하기

pymata4의 편리한 점 중 하나는 아두이노를 자동으로 찾아준다는 것이다. 아두이노와 호스트 PC간에 

정상적으로 연결만 되어있으면 단지 생성자 함수인 Pymata4()를 호출하는 것만으로 자동으로 

아두이노 보드를 찾아준다.

 

보드를 찾는 과정은 다음과 같은 로그로 확인할 수 있다.

 


/Volumes/Storage2/Firmata/FirmataTest/venv/bin/python /Volumes/Storage2/Firmata/FirmataTest/digital_write.py

pymata4: Version 1.10

 

Copyright (c) 2020 Alan Yorinks All Rights Reserved.

 

Opening all potential serial ports...

/dev/cu.wchusbserial1a1230

 

Waiting 4 seconds(arduino_wait) for Arduino devices to reset...

 

Searching for an Arduino configured with an arduino_instance = 1

Arduino compatible device found and connected to /dev/cu.wchusbserial1a1230

 

Retrieving Arduino Firmware ID...

Arduino Firmware ID: 1.1 FirmataExpress.ino

 

Retrieving analog map...

Auto-discovery complete. Found 22 Digital Pins and 8 Analog Pins


 

물론 명시적으로 필요한 파라미터를 전달하여 보드의 인스턴스를 생성할 수도 있다. 전달 가능한 파라미터는

다음과 같다(= 기호의 좌측은 파라미터 이름, 우측은 파라미터의 기본 값이다. Python은 파라미터 이름을 

사용하면 파라미터 순서와 관계없이 파라미터를 전달할 수 있으며 파라미터를 전달하지 않으면 = 우측의

기본 값이 사용된다).

 

  • com_port=None : 아두이노와 연결된 호스트PC의 포트. 위의 예에서는 /dev/cu.wchusbserial1a1230이다(매킨토시 PC에서 연결한 경우임).

  • baud_rate=115200 : 시리얼 통신 전송 속도

  • arduino_instance_id=1 : 아두이노 보드의 인스턴스 ID

  • arduino_wait=4 : 아두이노 리셋까지의 대기 시간

  • sleep_tune=0.000001 : 튜닝을 위한 파라미터로 일반적으로 수정하지 않는 것이 좋다.

  • shutdown_on_exception=True : RuntimeError Exception이나 KeyboardInterrupt exception이 발생한 경우 shutdown 함수를 호출할지의 여부

  • ip_address=None : StandardFirmataWifi 스케치를 사용하는 경우 Wi-Fi 기기의 IP 주소

  • ip_port=None : StandardFirmataWifi 스케치를 사용하는 경우 Wi-Fi 기기의 포트 번호. 보통 3030 사용

 

이와 같은 많은 파라미터들이 있으나 아무런 파라미터를 전달하지 않더라도 pymata4가 자동으로 연결된 

보드를 찾아 주는 것이다.

 

이 생성자 함수는 pymata4에서 사용할 전역 변수들을 초기화 하고 실제로 보드가 연결된 시리얼 포트를

찾는 것까지 상당히 긴 코드로 구현이 되어있어 자세한 설명은 생략하도록 하겠다. 다만 특별한 경우가 

아니라면  board = pymata4.Pymata4() 코드 한 줄로 아두이노를 연결하여 사용할 수 있다는 것만

알아두도록 하자.


blink 함수 분석 - 핀 모드 설정

Firmata를 이용하여 호스트 PC에서 아두이노의 핀을 이용하기 위해서는 먼저 아두이노 핀 모드를 설정해

주어야 한다. 그래서 가장 먼저 호출되는 API가 my_board.set_pin_mode_digital_output(pin)

이다. API 함수 이름으로 짐작할 수 있듯이 파라미터로 전달되는 pin(여기서는 6번 핀)을 디지털 출력으로 

설정하겠다는 의미이다. 그 과정을 조금 더 자세하게 살펴보자.

 

pymata4.py에서 set_pin_mode_digital_output 함수는 매우 단순하게 구현되어있다. 단지 전달

받은 파라미터인 pin에 private_constants.py에 정의된 상수인 OUPUT을 파라미터로 추가하여 private

함수인 _set_pin_mode 함수를 호출하는 것이 다이다. 전체 코드는 아래와 같다(원래의 주석은 삭제

했다).

 

def set_pin_mode_digital_output(self, pin_number):
        self._set_pin_mode(pin_number, PrivateConstants.OUTPUT)

 

_set_pin_mode 함수는 공통 함수이기 때문에 좀 더 복잡하게 구현이 되어있다. 하지만 우리는 Digital 

핀을 OUTPUT으로 사용할 것이므로 관련 부분만 집중해서 보도록 하자.

 

우선 전체 코드는 다음과 같다. 원래의 주석은 삭제를 하고 간단게 관련된 부분에만 별도의 주석을 달았다.

 

# Python에서 class의 멤버 함수들은 반드시 첫 번째 파라미터가 self여야 한다.
# 하지만 함수를 호출할 때는 self는 생략한다. 따라서 앞서 set_pin_mode_digital_output
# 함수에서 전달한 파라미터는 pin_number와 pin_state 2개이다. 나머지 2개의 파라미터는
# 전달받지 않았으므로 기본 값이 사용된다. 즉, callback은 None, differential은 1이다.
    def _set_pin_mode(self, pin_number, pin_state, callback=None,
                      differential=1):
# callback 파라미터는 전달받지 않았고 따라서 기본 값인 None을 사용하므로 아래 if문은
# 실행되지 않는다(None은 다른 언어의 null로 생각하면 된다).
        if callback:
            if pin_state == PrivateConstants.INPUT:
                self.digital_pins[pin_number].cb = callback
            elif pin_state == PrivateConstants.PULLUP:
                self.digital_pins[pin_number].cb = callback
                self.digital_pins[pin_number].pull_up = True
            elif pin_state == PrivateConstants.ANALOG:
                self.analog_pins[pin_number].cb = callback
                self.analog_pins[pin_number].differential = differential
            else:
                print('{} {}'.format('set_pin_mode: callback ignored for '
                                     'pin state:', pin_state))
#################################################################

# pin_state는 PrivateConstants.OUTPUT 값을 전달받았으므로 Pin_mode 역시
# PrivateConstants.OUTPUT 이다.
        pin_mode = pin_state

# pin_mode가 PrivateConstants.OUTPUT 이므로 아래 if문도 실행되지 않는다.
        if pin_mode == PrivateConstants.ANALOG:
            pin_number = pin_number + self.first_analog_pin
#################################################################

# _send_command 함수를 호출하여 아두이노로 통신하기 전에 마지막으로 Firmata 
# 프로토콜을 구현한다.
        command = [PrivateConstants.SET_PIN_MODE, pin_number, pin_mode]
        self._send_command(command)

# pin_mode가 PrivateConstants.OUTPUT 이므로 아래 if에서는 else절로 들어가
# 아무것도 실행하지 않고 끝낸다.
        if pin_state == PrivateConstants.INPUT or pin_state == PrivateConstants.PULLUP:
            self.enable_digital_reporting(pin_number)
        else:
            pass

 

 

위 소스 코드를 보면 대부분의 코드들은 파라미터 조건에 의해 실행되지 않거나 별다른 실행 없이 넘어가고

가장 핵심적인 부분인 Firmata 프로토콜을 구성하여 아두이노로 보내는 2줄만이 실행된다.

 

command = [PrivateConstants.SET_PIN_MODE, pin_number, pin_mode]
self._send_command(command)

 

Firmata 프로토콜 정의 문서(protocol/protocol.md at master · firmata/protocol · GitHub)

에서 보면 Message Type에 set pin mode(I/O) 라는 항목이 있고 이 항목은 다음과 같이 구성된다.

 

command MIDI channel
first byte second byte
0xF4   pin # (0~127) pin mode

 

 

그럼 이제 private_constants.py 파일을 열어 이 과정에서 사용된 2개의 상수인 OUTPUT과  SET_PIN_MODE

어떻게 정의되어 있는지 살펴보자.

 

SET_PIN_MODE = 0xF4  # set a pin to INPUT/OUTPUT/PWM/etc
...
OUTPUT = 0x01  # pin set as output

 

프로토콜로 정의된 것과 동일하게 command에 해당하는 SET_PIN_MODE0xF4 라는 값이 설정되어

있다. 나머지 2개의 파라미터는 범위 내에서 유동적으로 지정할 수 있으며 second byte에 해당하는 

OUTPUT은 카테고리 성격으로 자주 사용되는 값이므로 별도로 상수로 지정해 놓은 것이다.

 

마지막 단계의 _send_command함수도 단순한데, 전달받은 파라미터(프로토콜 포맷)를 byte 배열로 

바꾸어 serial port에 write(전송)하는 것이 끝이다. 전체 코드는 다음과 같다.

 

def _send_command(self, command):
        # 전달받은 파라미터(Firmata의 프로토콜 규격에 맞춘 데이터)를 바이트 배열로 바꾼다.
        send_message = bytes(command)

	# ip_address 전역변수가 지정되지 않은 경우 if 절 내의 내용을 실행한다.
        # 앞서 pymata4의 생성자 파라미터 설명에서 보았듯이 ip_address는 
        # StandardFirmataWifi라는 스케치에서만 사용되므로 여기서는 지정되지 않아
        # if 절이 실행되는 것이다.
        if not self.ip_address:
            try:
		# serial port에 write한다.
                result = self.serial_port.write(send_message)
            except SerialException:
                if self.shutdown_on_exception:
                    self.shutdown()
                raise RuntimeError('write fail in _send_command')
            return result
        else:
            self.sock.sendall(send_message)

 

정리

 

코드와 함께 정리하다보니 포스팅이 너무 길어진데다가 blink 함수에서 호출되는 digital_write API는 조금

복잡한 내용을 담고 있어서 다음 포스팅으로 넘기는 것이 좋을 것 같다.

 

다시 한번 강조하지만 전체 소스를 세세하게 파헤치는 것이 목표가 아니라 Firmata라는 프로토콜과 관련

라이브러리들이 어떤 흐름을 가지고 작동하는지를 아는 것이 목적인 만큼 관련 부분만 주목해서 파악하면

될 것 같다. 하지만 소프트웨어라는 것이 어찌보면 상당히 유기적이기도 한 만큼 생략으로 인해 혼란 스러운

부분도 많을 것이다. 이러한 부분들은 추후 기회를 봐서 더 상세히 다뤄보도록 하겠다.

 

그럼 다음 포스팅에서 digital_write API에 대한 내용으로 이어가겠다.

 

반응형