今回の記事ではUSBカメラでPythonとOpencvを使用してQR Codeを読みこみます。そしてQR Codeの結果をDocker上にあるRedis Serverに保存し、またModbus TCP経由で三菱FX5Uから撮影のトリガーとQR Codeの内容を転送するようにします。

さ、FAを楽しもう。

Reference Link

http://soup01.com/ja/category/database/redis/

PhoenixContact#AXL F BK ETH EFを使ってみよう

EXOR#Part16_Modbus TCP Clientを使ってみよう

Implementation

Python Side

最初にSeeed Stuidoのrecomputer R1025-10にPython Scriptを書きます。

Install packages

まずpipを使用しPythonパッケージをインストールします。

redis

こちらのコマンドでredis ServerにアクセスするPython Packagesをインストールします。

https://redis.io/docs/latest/develop/clients/redis-py/

pip install redis

こちらはredis Packageを使用しredis Serverをアクセスするサンプルコードになります。

import redis

#
host=’192.168.13.159′
port=6379
decode_responses=True
r=redis.Redis(host=host,port=port,decode_responses=decode_responses)
hash_key=”qr_data”

r.hset(hash_key,”2024-07-27-2″,”www.abc..jp1″)

redis_qrhashkey=’qr_data’

opencv

こちらのコマンドでOpencv のPython Packagesをインストールします。

https://pypi.org/project/opencv-python/

pip install opencv-python

pymodbus

こちらのコマンドでModbus TCP 通信のPython Packagesをインストールします。

https://pymodbus.readthedocs.io/en/latest/

Scripts

こちらは今回記事で作成したScriptsです。

import cv2
import struct
from pymodbus.client import ModbusTcpClient
from datetime import datetime
import signal
import sys
from time import sleep
import redis
def signal_handler(sig, frame):
print(‘You pressed Ctrl+C!’)
client.close()
print(“Client is closed..”)
sys.exit(0)

def capture_photo_from_webcam(save_path):
cap = cv2.VideoCapture(0)
if not cap.isOpened():
print(“Cannot open camera”)
return
ret,frame=cap.read()

if not ret:
print(“Cannot grab frame”)
else:
cv2.imwrite(save_path, frame)
print(f”Image saved as {save_path}”)
cap.release()

server_ip=”127.0.0.1″
server_port=11502redis_host=’192.168.13.159’redis_port=6379redis_decode_responses=Trueredis_qrhashkey=’qr_data’
client = ModbusTcpClient(server_ip, port=server_port)
r=redis.Redis(host=redis_host,port=redis_port,decode_responses=redis_decode_responses)
triggered=False
register_backup=0
register=0
registers = []
zero_data = [0] * 99
signal.signal(signal.SIGINT, signal_handler)
if client.connect():
while True:
sleep(0.05)
result=client.read_holding_registers(200,1)
now=datetime.now()
dt_string = now.strftime(“%d-%m-%Y-%H-%M-%S”)
image_path=(‘/home/recomputer/Pictures/captured_’+dt_string+’_image.png’)
if not result.isError():
print(“Register:”,result.registers[0])
register=result.registers[0]
if result.registers[0] != register_backup:
register_backup=result.registers[0]
if result.registers[0] ==1:
triggered=True
print(“triggerred..”)
response = client.write_registers(1,[2])
else:
print(“Error..”)
response = client.write_registers(1,[90])
if triggered:
client.write_registers(2, zero_data)
print(“—Reset all registers..—“)
capture_photo_from_webcam(image_path)
print(“–Capature..–“)
now=datetime.now()
print(“–Read Images..–“)
image = cv2.imread(image_path)
detector = cv2.QRCodeDetector()
data, vertices_array, _ = detector.detectAndDecode(image)

if vertices_array is not None:
print(data)
response = client.write_registers(1,[3])
else:
data=”ERROR”
response = client.write_registers(1,[91])
registers=[] 　r.hset(redis_qrhashkey,dt_string,data)
byte_data = data.encode(‘ascii’)
for i in range(0, len(byte_data), 2):
two_bytes = byte_data[i:i+2]
if len(two_bytes) == 1:
two_bytes += b’\x00′
registers.append(struct.unpack(‘>H’, two_bytes)[0])

response = client.write_registers(2, registers)
print(“—-“)
triggered=False
client.close()
print(“Disconnect..”)

Import librarys

最初はOpencv・Modbus・Redisなど必要なライブラリをImportします。

import cv2import structfrom pymodbus.client import ModbusTcpClientfrom datetime import datetimeimport signalimport sysfrom time import sleepimport redis

Handle Ctrl+C

こちらはTerminalでCtrlx+Cを押したときの処理になります。

def signal_handler(sig, frame):
print(‘You pressed Ctrl+C!’)
client.close()
print(“Client is closed..”)
sys.exit(0)

Capture WebCam Frame

こちらはOpencvライブラリを使用しWebカメラをトリガーし、写真を撮影するプログラムです。

def capture_photo_from_webcam(save_path):
cap = cv2.VideoCapture(0)
if not cap.isOpened():
print(“Cannot open camera”)
return
ret,frame=cap.read()

if not ret:
print(“Cannot grab frame”)
else:
cv2.imwrite(save_path, frame)
print(f”Image saved as {save_path}”)
cap.release()

Configure ModbusTCP/redis Server Connection

こちらはModbus TCP Serverとredis Data Serverの接続設定になります。

server_ip=”127.0.0.1″server_port=11502redis_host=’192.168.13.159’redis_port=6379redis_decode_responses=Trueredis_qrhashkey=’qr_data’
client = ModbusTcpClient(server_ip, port=server_port)r=redis.Redis(host=redis_host,port=redis_port,decode_responses=redis_decode_responses)

Init the variables

Modbus TCP Serverに読み書きするレジスタを初期化します。

triggered=False
register_backup=0
register=0
registers = []
zero_data = [0] * 99
signal.signal(signal.SIGINT, signal_handler)

Read Holding Registers

Modbus TCP Serverに接続成功した場合、Modbus TCP ServerにHolding レジスタ値を取得します。

if client.connect():
while True:
sleep(0.05)
result=client.read_holding_registers(200,1)
now=datetime.now()
dt_string = now.strftime(“%d-%m-%Y-%H-%M-%S”)

Configure Path

画像の保存先を生成します。

image_path=(‘/home/recomputer/Pictures/captured_’+dt_string+’_image.png’)

Get Data via Modbus Result if OK

Modbus TCP Serverのレジスタ読み成功の場合、該当するレジスタ=1なら写真の撮影動作をトリガーします。

if not result.isError():
print(“Register:”,result.registers[0])
register=result.registers[0]
if result.registers[0] != register_backup:
register_backup=result.registers[0]
if result.registers[0] ==1:
triggered=True
print(“triggerred..”)
response = client.write_registers(1,[2])
else:
print(“Error..”)
response = client.write_registers(1,[90])

Trigger the capture method

capture_photo_from_webcamのMethodを実行し写真を撮影し、QRCodeDetectorでQR Codeを取り出します。

if triggered:
client.write_registers(2, zero_data)
print(“—Reset all registers..—“)
capture_photo_from_webcam(image_path)
print(“–Capature..–“)
now=datetime.now()
dt_string = now.strftime(“%d/%m/%Y %H:%M:%S”)
print(“–Read Images..–“)
image = cv2.imread(image_path)
detector = cv2.QRCodeDetector()
data, vertices_array, _ = detector.detectAndDecode(image)

Transfer the Result via modbus

QR Codeの読み結果をModbus TCP Serverに書き込みます。

if vertices_array is not None:
print(data)
response = client.write_registers(1,[3])
else:
data=”ERROR”
response = client.write_registers(1,[91]) r.hset(redis_qrhashkey,dt_string,data)
registers=[]
byte_data = data.encode(‘ascii’)
for i in range(0, len(byte_data), 2):
two_bytes = byte_data[i:i+2]
if len(two_bytes) == 1:
two_bytes += b‘\x00’
registers.append(struct.unpack(‘>H’, two_bytes)[0])

response = client.write_registers(2, registers)
print(“—-“)
triggered=False