【Python】Raspberry Piで赤外線受信

2020年12月31日2021年1月2日

pigpioを使った赤外線受信をPythonで実装してみたので、自分用に内容を記事にしときたいと思います。

前回の記事で赤外線受信を動かす為の配線を行いましたので、これを動かしていきます。

Raspberry Piで赤外線受信を実装

今回はラズパイZeroを使用し、赤外線受信をできるようにした時の内容を備忘録として残しておきます。 ...

https://nompor.com/2020/12/31/post-6352/

http://abyz.me.uk/rpi/pigpio/code/irrp_py.zipの赤外線受信のPythonコードを流用して赤外線受信専用にしてます。

プログラム上で指定するのはピン番号ではなくGPIO番号です。

ラズパイのGPIO番号

プログラムはリモコンの赤外線発信を待機し、受信したら配列のJSONテキストをファイルとして書き出します。

起動して待機状態になったら、リモコンを受信部分に向けてボタンを押します。

処理が完了したらOkayのログが表示されます。

import time
import json
import os
import argparse
import pigpio

GPIO       = 17
FILE       = "recv.txt"
GLITCH     = 100
PRE_MS     = 200
POST_MS    = 150
FREQ       = 38
VERBOSE    = False
SHORT      = 10
GAP_MS     = 100
TOLERANCE  = 15

POST_US    = POST_MS * 1000
PRE_US     = PRE_MS  * 1000
GAP_S      = GAP_MS  / 1000.0
TOLER_MIN =  (100 - TOLERANCE) / 100.0
TOLER_MAX =  (100 + TOLERANCE) / 100.0

last_tick = 0
in_code = False
code = []
fetching_code = False

def normalise(c):
   if VERBOSE:
      print("before normalise", c)
   entries = len(c)
   p = [0]*entries # Set all entries not processed.
   for i in range(entries):
      if not p[i]: # Not processed?
         v = c[i]
         tot = v
         similar = 1.0

         # Find all pulses with similar lengths to the start pulse.
         for j in range(i+2, entries, 2):
            if not p[j]: # Unprocessed.
               if (c[j]*TOLER_MIN) < v < (c[j]*TOLER_MAX): # Similar.
                  tot = tot + c[j]
                  similar += 1.0

         # Calculate the average pulse length.
         newv = round(tot / similar, 2)
         c[i] = newv

         # Set all similar pulses to the average value.
         for j in range(i+2, entries, 2):
            if not p[j]: # Unprocessed.
               if (c[j]*TOLER_MIN) < v < (c[j]*TOLER_MAX): # Similar.
                  c[j] = newv
                  p[j] = 1

   if VERBOSE:
      print("after normalise", c)

def compare(p1, p2):
   if len(p1) != len(p2):
      return False

   for i in range(len(p1)):
      v = p1[i] / p2[i]
      if (v < TOLER_MIN) or (v > TOLER_MAX):
         return False

   for i in range(len(p1)):
       p1[i] = int(round((p1[i]+p2[i])/2.0))

   if VERBOSE:
      print("after compare", p1)

   return True

def tidy_mark_space(records, base):

   ms = {}

   # Find all the unique marks (base=0) or spaces (base=1)
   # and count the number of times they appear,

   rl = len(records)
   for i in range(base, rl, 2):
      if records[i] in ms:
         ms[records[i]] += 1
      else:
         ms[records[i]] = 1

   if VERBOSE:
      print("t_m_s A", ms)

   v = None

   for plen in sorted(ms):

      if v == None:
         e = [plen]
         v = plen
         tot = plen * ms[plen]
         similar = ms[plen]

      elif plen < (v*TOLER_MAX):
         e.append(plen)
         tot += (plen * ms[plen])
         similar += ms[plen]

      else:
         v = int(round(tot/float(similar)))
         # set all previous to v
         for i in e:
            ms[i] = v
         e = [plen]
         v = plen
         tot = plen * ms[plen]
         similar = ms[plen]

   v = int(round(tot/float(similar)))
   # set all previous to v
   for i in e:
      ms[i] = v

   if VERBOSE:
      print("t_m_s B", ms)

   rl = len(records)
   for i in range(base, rl, 2):
      records[i] = ms[records[i]]

def tidy(records):

   tidy_mark_space(records, 0) # Marks.

   tidy_mark_space(records, 1) # Spaces.

def end_of_code():
   global code, fetching_code
   if len(code) > SHORT:
      normalise(code)
      fetching_code = False
   else:
      code = []
      print("Short code, probably a repeat, try again")

def cbf(gpio, level, tick):
   global last_tick, in_code, code, fetching_code



   if level != pigpio.TIMEOUT:

      edge = pigpio.tickDiff(last_tick, tick)
      last_tick = tick

      if fetching_code:

         if (edge > PRE_US) and (not in_code): # Start of a code.
            in_code = True
            pi.set_watchdog(GPIO, POST_MS) # Start watchdog.

         elif (edge > POST_US) and in_code: # End of a code.
            in_code = False
            pi.set_watchdog(GPIO, 0) # Cancel watchdog.
            end_of_code()

         elif in_code:
            code.append(edge)

   else:
      pi.set_watchdog(GPIO, 0) # Cancel watchdog.
      if in_code:
         in_code = False
         end_of_code()


pi = pigpio.pi() # Connect to Pi.

if not pi.connected:
   exit(0)

try:
  f = open(FILE, "r")
  records = json.load(f)
  f.close()
except:
  records = []

pi.set_mode(GPIO, pigpio.INPUT) # IR RX connected to this GPIO.

pi.set_glitch_filter(GPIO, GLITCH) # Ignore glitches.

cb = pi.callback(GPIO, pigpio.EITHER_EDGE, cbf)

# Process each id

print("Recording")
print("Press remote control key")
code = []
fetching_code = True
while fetching_code:
  time.sleep(0.1)
print("Okay")
time.sleep(0.5)

records = code[:]

pi.set_glitch_filter(GPIO, 0) # Cancel glitch filter.
pi.set_watchdog(GPIO, 0) # Cancel watchdog.

tidy(records)

f = open(FILE, "w")
f.write(json.dumps(records, sort_keys=True).replace("],", "],\n")+"\n")
f.close()

pi.stop() # Disconnect from Pi.

実行結果