라즈베리파이 미세먼지 센서(PMS7003)

Posted on: 2019년 10월 02일 2019년 10월 02일
Categories: 강의자료, 라즈베리파이

참조
1. PMS7003 활용 예 : http://blog.naver.com/elepartsblog/221347040698
필요 장비
1. PMS7003 먼지 센서
2. 먼지센서 인터페이스 보드
3. USB to UART 변환 케이블
시리얼 통신을 위한 설정 : 참조
데이터 시트 : http://eleparts.co.kr/data/_gextends/good-pdf/201803/good-pdf-4208690-1.pdf
출력 모드 2가지 존재
1. Active Mode : 기본 설정, 자동으로 측정값 전송, 예제 프로그램은 Activie Mode용
  1. stable mode : 2.3s 간격
  2. fast mode : 200~800ms 간격
2. Passive Mode : 요청이 있을 때만 측정값 전송, 재작동시 30초 이후에 값 측정해야 정확

샘플코드 : https://github.com/eleparts/PMS7003

"""
* PMS7003 데이터 수신 프로그램
* 수정 : 2018. 11. 19
* 제작 : eleparts 부설연구소
* SW ver. 1.0.2

> 관련자료
파이썬 라이브러리
https://docs.python.org/3/library/struct.html

점프 투 파이썬
https://wikidocs.net/book/1

PMS7003 datasheet
http://eleparts.co.kr/data/_gextends/good-pdf/201803/good-pdf-4208690-1.pdf
"""
import serial
import struct
import time
from datetime import datetime


class PMS7003(object):

  # PMS7003 protocol data (HEADER 2byte + 30byte)
  PMS_7003_PROTOCOL_SIZE = 32

  # PMS7003 data list
  HEADER_HIGH            = 0  # 0x42
  HEADER_LOW             = 1  # 0x4d
  FRAME_LENGTH           = 2  # 2x13+2(data+check bytes)
  DUST_PM1_0_CF1         = 3  # PM1.0 concentration unit μ g/m3（CF=1，standard particle）
  DUST_PM2_5_CF1         = 4  # PM2.5 concentration unit μ g/m3（CF=1，standard particle）
  DUST_PM10_0_CF1        = 5  # PM10 concentration unit μ g/m3（CF=1，standard particle）
  DUST_PM1_0_ATM         = 6  # PM1.0 concentration unit μ g/m3（under atmospheric environment）
  DUST_PM2_5_ATM         = 7  # PM2.5 concentration unit μ g/m3（under atmospheric environment）
  DUST_PM10_0_ATM        = 8  # PM10 concentration unit μ g/m3  (under atmospheric environment)
  DUST_AIR_0_3           = 9  # indicates the number of particles with diameter beyond 0.3 um in 0.1 L of air.
  DUST_AIR_0_5           = 10 # indicates the number of particles with diameter beyond 0.5 um in 0.1 L of air.
  DUST_AIR_1_0           = 11 # indicates the number of particles with diameter beyond 1.0 um in 0.1 L of air.
  DUST_AIR_2_5           = 12 # indicates the number of particles with diameter beyond 2.5 um in 0.1 L of air.
  DUST_AIR_5_0           = 13 # indicates the number of particles with diameter beyond 5.0 um in 0.1 L of air.
  DUST_AIR_10_0          = 14 # indicates the number of particles with diameter beyond 10 um in 0.1 L of air.
  RESERVEDF              = 15 # Data13 Reserved high 8 bits
  RESERVEDB              = 16 # Data13 Reserved low 8 bits
  CHECKSUM               = 17 # Checksum code


  # header check
  def header_chk(self, buffer):
    if (buffer[self.HEADER_HIGH] == 66 and buffer[self.HEADER_LOW] == 77):
      return True
    else:
      return False


  # chksum value calculation
  def chksum_cal(self, buffer):
    buffer = buffer[0:self.PMS_7003_PROTOCOL_SIZE]
    # data unpack (Byte -> Tuple (30 x unsigned char <B> + unsigned short <H>))
    chksum_data = struct.unpack('!30BH', buffer)
    chksum = 0

    for i in range(30):
      chksum = chksum + chksum_data[i]

    return chksum


  # checksum check
  def chksum_chk(self, buffer):
    chk_result = self.chksum_cal(buffer)
    chksum_buffer = buffer[30:self.PMS_7003_PROTOCOL_SIZE]
    chksum = struct.unpack('!H', chksum_buffer)

    if (chk_result == chksum[0]):
      return True
    else:
      return False


  # protocol size(small) check
  def protocol_size_chk(self, buffer):
    if(self.PMS_7003_PROTOCOL_SIZE <= len(buffer)):
      return True
    else:
      return False


  # protocol check
  def protocol_chk(self, buffer):
    if(self.protocol_size_chk(buffer)):
      if(self.header_chk(buffer)):
        if(self.chksum_chk(buffer)):
          return True
        else:
          print("Chksum err")
      else:
        print("Header err")
    else:
      print("Protol err")

    return False

  # unpack data
  # <Tuple (13 x unsigned short <H> + 2 x unsigned char <B> + unsigned short <H>)>
  def unpack_data(self, buffer):
    buffer = buffer[0:self.PMS_7003_PROTOCOL_SIZE]
    # data unpack (Byte -> Tuple (13 x unsigned short <H> + 2 x unsigned char <B> + unsigned short <H>))
    data = struct.unpack('!2B13H2BH', buffer)

    return data


  def print_serial(self, buffer):
    chksum = self.chksum_cal(buffer)
    data = self.unpack_data(buffer)

    print ("============================================================================")
    print ("Header : %c %c \t\t | Frame length : %s" % (data[self.HEADER_HIGH], data[self.HEADER_LOW], data[self.FRAME_LENGTH]))
    print ("PM 1.0 (CF=1) : %s\t | PM 1.0 : %s" % (data[self.DUST_PM1_0_CF1], data[self.DUST_PM1_0_ATM]))
    print ("PM 2.5 (CF=1) : %s\t | PM 2.5 : %s" % (data[self.DUST_PM2_5_CF1], data[self.DUST_PM2_5_ATM]))
    print ("PM 10.0 (CF=1) : %s\t | PM 10.0 : %s" % (data[self.DUST_PM10_0_CF1], data[self.DUST_PM10_0_ATM]))
    print ("0.3um in 0.1L of air : %s" % (data[self.DUST_AIR_0_3]))
    print ("0.5um in 0.1L of air : %s" % (data[self.DUST_AIR_0_5]))
    print ("1.0um in 0.1L of air : %s" % (data[self.DUST_AIR_1_0]))
    print ("2.5um in 0.1L of air : %s" % (data[self.DUST_AIR_2_5]))
    print ("5.0um in 0.1L of air : %s" % (data[self.DUST_AIR_5_0]))
    print ("10.0um in 0.1L of air : %s" % (data[self.DUST_AIR_10_0]))
    print ("Reserved F : %s | Reserved B : %s" % (data[self.RESERVEDF],data[self.RESERVEDB]))
    print ("CHKSUM : %s | read CHKSUM : %s | CHKSUM result : %s" % (chksum, data[self.CHECKSUM], chksum == data[self.CHECKSUM]))
    print ("============================================================================")



# UART / USB Serial : 'dmesg | grep ttyUSB'
USB0 = '/dev/ttyUSB0'
UART = '/dev/ttyAMA0'

# USE PORT
SERIAL_PORT = USB0

# Baud Rate
Speed = 9600


# example
if __name__=='__main__':
  #serial setting
  ser = serial.Serial(SERIAL_PORT, Speed, timeout = 5)
  dust = PMS7003()

  while True:
    ser.flushInput()
    buffer = ser.read(1024)

    if(dust.protocol_chk(buffer)):
      now = datetime.now()
      print("DATA read success %02d:%02d:%02d" %(now.hour, now.minute, now.second))
      # print data
      dust.print_serial(buffer)
    else:
      print("DATA read fail...")

  ser.close()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

"""

* PMS7003 데이터 수신 프로그램

* 수정 : 2018. 11. 19

* 제작 : eleparts 부설연구소

* SW ver. 1.0.2

> 관련자료

파이썬 라이브러리

https://docs.python.org/3/library/struct.html

점프 투 파이썬

https://wikidocs.net/book/1

PMS7003 datasheet

http://eleparts.co.kr/data/_gextends/good-pdf/201803/good-pdf-4208690-1.pdf

"""

import serial

import struct

import time

from datetime import datetime

class PMS7003(object):

# PMS7003 protocol data (HEADER 2byte + 30byte)

PMS_7003_PROTOCOL_SIZE = 32

# PMS7003 data list

HEADER_HIGH = 0 # 0x42

HEADER_LOW = 1 # 0x4d

FRAME_LENGTH = 2 # 2x13+2(data+check bytes)

DUST_PM1_0_CF1 = 3 # PM1.0 concentration unit μ g/m3（CF=1，standard particle）

DUST_PM2_5_CF1 = 4 # PM2.5 concentration unit μ g/m3（CF=1，standard particle）

DUST_PM10_0_CF1 = 5 # PM10 concentration unit μ g/m3（CF=1，standard particle）

DUST_PM1_0_ATM = 6 # PM1.0 concentration unit μ g/m3（under atmospheric environment）

DUST_PM2_5_ATM = 7 # PM2.5 concentration unit μ g/m3（under atmospheric environment）

DUST_PM10_0_ATM = 8 # PM10 concentration unit μ g/m3 (under atmospheric environment)

DUST_AIR_0_3 = 9 # indicates the number of particles with diameter beyond 0.3 um in 0.1 L of air.

DUST_AIR_0_5 = 10 # indicates the number of particles with diameter beyond 0.5 um in 0.1 L of air.

DUST_AIR_1_0 = 11 # indicates the number of particles with diameter beyond 1.0 um in 0.1 L of air.

DUST_AIR_2_5 = 12 # indicates the number of particles with diameter beyond 2.5 um in 0.1 L of air.

DUST_AIR_5_0 = 13 # indicates the number of particles with diameter beyond 5.0 um in 0.1 L of air.

DUST_AIR_10_0 = 14 # indicates the number of particles with diameter beyond 10 um in 0.1 L of air.

RESERVEDF = 15 # Data13 Reserved high 8 bits

RESERVEDB = 16 # Data13 Reserved low 8 bits

CHECKSUM = 17 # Checksum code

# header check

def header_chk(self, buffer):

if (buffer[self.HEADER_HIGH] == 66 and buffer[self.HEADER_LOW] == 77):

return True

else:

return False

# chksum value calculation

def chksum_cal(self, buffer):

buffer = buffer[0:self.PMS_7003_PROTOCOL_SIZE]

# data unpack (Byte -> Tuple (30 x unsigned char <B> + unsigned short <H>))

chksum_data = struct.unpack('!30BH', buffer)

chksum = 0

for i in range(30):

chksum = chksum + chksum_data[i]

return chksum

# checksum check

def chksum_chk(self, buffer):

chk_result = self.chksum_cal(buffer)

chksum_buffer = buffer[30:self.PMS_7003_PROTOCOL_SIZE]

chksum = struct.unpack('!H', chksum_buffer)

if (chk_result == chksum[0]):

return True

else:

return False

# protocol size(small) check

def protocol_size_chk(self, buffer):

if(self.PMS_7003_PROTOCOL_SIZE <= len(buffer)):

return True

else:

return False

# protocol check

def protocol_chk(self, buffer):

if(self.protocol_size_chk(buffer)):

if(self.header_chk(buffer)):

if(self.chksum_chk(buffer)):

return True

else:

print("Chksum err")

else:

print("Header err")

else:

print("Protol err")

return False

# unpack data

# <Tuple (13 x unsigned short <H> + 2 x unsigned char <B> + unsigned short <H>)>

def unpack_data(self, buffer):

buffer = buffer[0:self.PMS_7003_PROTOCOL_SIZE]

# data unpack (Byte -> Tuple (13 x unsigned short <H> + 2 x unsigned char <B> + unsigned short <H>))

data = struct.unpack('!2B13H2BH', buffer)

return data

def print_serial(self, buffer):

chksum = self.chksum_cal(buffer)

data = self.unpack_data(buffer)

print ("============================================================================")

print ("Header : %c %c \t\t | Frame length : %s" % (data[self.HEADER_HIGH], data[self.HEADER_LOW], data[self.FRAME_LENGTH]))

print ("PM 1.0 (CF=1) : %s\t | PM 1.0 : %s" % (data[self.DUST_PM1_0_CF1], data[self.DUST_PM1_0_ATM]))

print ("PM 2.5 (CF=1) : %s\t | PM 2.5 : %s" % (data[self.DUST_PM2_5_CF1], data[self.DUST_PM2_5_ATM]))

print ("PM 10.0 (CF=1) : %s\t | PM 10.0 : %s" % (data[self.DUST_PM10_0_CF1], data[self.DUST_PM10_0_ATM]))

print ("0.3um in 0.1L of air : %s" % (data[self.DUST_AIR_0_3]))

print ("0.5um in 0.1L of air : %s" % (data[self.DUST_AIR_0_5]))

print ("1.0um in 0.1L of air : %s" % (data[self.DUST_AIR_1_0]))

print ("2.5um in 0.1L of air : %s" % (data[self.DUST_AIR_2_5]))

print ("5.0um in 0.1L of air : %s" % (data[self.DUST_AIR_5_0]))

print ("10.0um in 0.1L of air : %s" % (data[self.DUST_AIR_10_0]))

print ("Reserved F : %s | Reserved B : %s" % (data[self.RESERVEDF],data[self.RESERVEDB]))

print ("CHKSUM : %s | read CHKSUM : %s | CHKSUM result : %s" % (chksum, data[self.CHECKSUM], chksum == data[self.CHECKSUM]))

print ("============================================================================")

# UART / USB Serial : 'dmesg | grep ttyUSB'

USB0 = '/dev/ttyUSB0'

UART = '/dev/ttyAMA0'

# USE PORT

SERIAL_PORT = USB0

# Baud Rate

Speed = 9600

# example

if __name__=='__main__':

#serial setting

ser = serial.Serial(SERIAL_PORT, Speed, timeout = 5)

dust = PMS7003()

while True:

ser.flushInput()

buffer = ser.read(1024)

if(dust.protocol_chk(buffer)):

now = datetime.now()

print("DATA read success %02d:%02d:%02d" %(now.hour, now.minute, now.second))

# print data

dust.print_serial(buffer)

else:

print("DATA read fail...")

ser.close()

PMS7003 센서에서 전송되는 값
1. 32바이트로 구성된 18개의 값
  1. 1바이트 : 0x42
  2. 1바이트 : 0x4d
  3. 2바이트 : 28
  4. 2바이트 : PM1.0 concentration unit μ g/m3（CF=1，standard particle）
  5. 2바이트 : PM2.5 concentration unit μ g/m3（CF=1，standard particle）
  6. 2바이트 : PM10 concentration unit μ g/m3（CF=1，standard particle）
  7. 2바이트 : PM1.0 concentration unit μ g/m3（under atmospheric environment）
  8. 2바이트 : PM2.5 concentration unit μ g/m3（under atmospheric environment）
  9. 2바이트 : PM10 concentration unit μ g/m3 (under atmospheric environment)
  10. 2바이트 : indicates the number of particles with diameter beyond 0.3 um in 0.1 L of air.
  11. 2바이트 : indicates the number of particles with diameter beyond 0.5 um in 0.1 L of air.
  12. 2바이트 : indicates the number of particles with diameter beyond 1.0 um in 0.1 L of air.
  13. 2바이트 : indicates the number of particles with diameter beyond 2.5 um in 0.1 L of air.
  14. 2바이트 : indicates the number of particles with diameter beyond 5.0 um in 0.1 L of air.
  15. 2바이트 : indicates the number of particles with diameter beyond 10 um in 0.1 L of air.
  16. 1바이트 : Data13 Reserved high 8 bits
  17. 1바이트 : Data13 Reserved low 8 bits
  18. 2바이트 : Checksum code
struct 모듈 : C의 union과 유사하게 bytes를 struct로 또는 그 반대로 상호 변환해 줌
1. 참조 : https://m.blog.naver.com/s2kiess/220243476924
2. struct.unpack(format, buffer) : byte로 구성된 buffer의 내용을 format 형태로 변환한 후 파이썬의 투플(tuple)로 표현됨
3. chksum_data = struct.unpack(‘!30BH’, buffer)
  1. bytes로 구성된 buffer를 30개의 B와 1개의 H로 변환
  2. ! : 정렬 제어문자 중 하나로 네트워크 바이트 순서(빅 엔디안)로 전송
    1. 리틀 엔디안 : 역순으로 저장되고 역순으로 읽어옴
    2. 빅 엔디안 : 순서대로 저장하고 순서대로 읽어옴
  3. B : unsigned char(1byte) -> 1바이트 정수
  4. H : unsigned short(2bytes) -> 2바이트 정수
  5. chksum_data = (B, B, B, … , B, H)
4. data = struct.unpack(‘!2B13H2BH’, buffer)
  1. bytes로 구성된 buffer를 2개의 B와 13개의 H, 2개의 B, 1개의 H로 변환
  2. data = (B, B, H, H, …, H, B, B, H)
5. struct.pack(format, v1, v2, …) : 여려개의 변수의 값들을 byte형태로 변환
6. struct.pack(‘hhl’, 1, 2, 3) => b’\x00\x01\x00\x02\x00\x00\x00\x03′
7. struct.pack(‘ci’, b’*’, 0x12131415) => b’*\x00\x00\x00\x12\x13\x14\x15‘
  1. 4바이트 단위로 저장되기 위해 문자(*) 뒤에 3바이트를 비운 듯 ?
8. struct.pack(‘ic’, 0x12131415, b’*’) => b’\x12\x13\x14\x15*‘
9. struct.calcsize(‘ci’) => 8
10. struct.calcsize(‘ic’) => 5
11. pack(‘llh0l’, 1, 2, 3) => b’\x00\x00\x00\x01\x00\x00\x00\x02\x00\x03\x00\x00‘

whan0623

517