ラズパイPythonと三菱PLCのネットワーク連携【イーサネットユニット編】-ICHIKEN@エンジニアキャンパーのバエないアメリカ生活

この記事ではラズパイと三菱Qシリーズイーサネットユニットとをネットワーク経由で連携させる方法を紹介します。QシリーズCPUユニットのイーサネットポートの場合の設定とテストはこちら。

ラズパイPythonと三菱PLCをネットワーク経由で連携させる方法

CONTENTS [非表示]

1この記事でわかること
2前提条件
3通信設定
- 3-1イーサネットユニットパラメータ設定
- 3-2ラズパイのIP設定
4pymcprotocolのインストール
5pymcprotocolを使った通信テスト
6テスト結果
7Pythonコードの解説
- 7-1流れ
8最後に

この記事でわかること

ラズパイと三菱Qシリーズイーサネットユニットとをネットワーク接続し、ラズパイ上のPythonからpymcprotocolを使ってPLCのIOを読み取る方法がわかります。

三菱PLC設定方法
ラズパイIP設定方法
Pythonライブラリ pymcprotocolインストール方法
pymcprotocolを使った通信方法

前提条件

まず前提条件となる私の環境です。

Raspberry Pi	4 Model B 8GB
Raspberry Pi OS	Raspbian GNU/Linux 10 (buster)
VSCode	1.55.2
Python	3.7.3
pymcprotocol	0.3.0
三菱PLC	Q03UDECPU CPU-LANポートは不使用
イーサネットユニット	QJ71E71-100
GX Works2	1.586L

ラズパイのPython開発環境はVisual Studio Codeを使用します。インストールはこちらを参照してください。

【IoT×キャンプ｜Python開発環境VSCode編】意地でもラズパイとキャンプへ！らずキャン△プロジェクト第6回

通信設定

イーサネットユニットパラメータ設定

イーサネットユニットのIPアドレスはローカル192.168.1.2を使用します。

イーサネットユニットにはCPUユニットと違いMCプロトコルの設定がないため別の設定を使用します。pymcprotocolのページにTCPを使えとあるのでTCPにします。受信ポートの指定がないようですのでポートのオープン設定はUnpassiveのみが使えそうです。

ほかにもいくつか設定があるため４パターンでどれが使えるかテストしてみます。

テストに使用するIPとポート

PLCのIPアドレス	192.168.1.2
ポート	TCP 1025 ~ 1028

以下の4パターンでテストしました。結果から言うと、1と2はデータ取得と書き込みができて、3と4はだめでした。1番を使っておけばOKです。

設定1：TCP、Unpassive、受信、手順あり
設定2：TCP、Unpassive、送信、手順あり
設定3：TCP、Unpassive、受信、無手順
設定4：TCP、Unpassive、受信、事前定義プロトコル

では、設定をサクサク解説します。まずGX Works2でパラメータ設定を開きIO割付をします。今回はとりあえず開始を0010にします。

次にネットワークパラメータ設定画面を開きます。IO割付を先程設定した0010にします。他の設定は今回は用が無いので全部1。

「動作設定」を開きIPアドレスなどを設定します。交信データコードは今回はあえてASCIIで行きます。バイナリの場合はPython側で指定は不要、ASCIIはPython側で指定が必要です。

次に「オープン設定」でプロトコルの設定をします。

設定が完了したら書き込んでPLCを再起動してください。再起動しないとパラメータ設定が反映されません。

ラズパイのIP設定

ラズパイのIPを192.168.1.3に設定します。

ラズパイのIPアドレス	192.168.1.3

pymcprotocolのインストール

PythonでMCプロトコルを扱うためのライブラリをインストールします。LXTerminalをひらいて次のコマンドを実行してください。

pip install pymcprotocol

インストールは以上です。ドキュメントはこちら。

Welcome to pymcprotocol's documentation! - pymcprotocol '0.3.0' documentation

LXTerminalでPythonを立ち上げてimportすることで動作確認できます。

 python3
import pymcprotocol

import後にエラーがなければインストールに問題はありません。

pymcprotocolを使った通信テスト

テスト準備

準備できましたので実際にテストしていきます。ラズパイとPLCをLANケーブルで直接接続しましょう。

テストの動作パターンですが、今回はPLCでY0をONするとラズパイがPLCのX0、Y0をOFFするとX0をOFFするようにします。モニター用にX0をM1出力するようにしています。

ラダー記述

テスト用のラダーを記述します。出力Y0をON/OFF繰り返す回路です。とりあえず1秒ON、2秒OFFにします。

Pythonコード記述

今回は2つのpyファイルで動くようにしています。mcp_test.pyとlog.pyを作り、同じフォルダ内に置いてmcp_test.pyを実行するのが最も簡単です。

mcp_test.py：メイン通信用。使い回しを念頭に置き、スレッドクラス化してます。
log.py：ログ出力用。

まずは「mcp_test.py」。こちらはファイル名は何でもよいです。

 #!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymcprotocol
import threading
import log # 自作モジュール
from logging import getLogger
import time
 
 
class InspectionThread(threading.Thread):
 
    def __init__(self):
 
        # ---- Setting for threading
        super(InspectionThread, self).__init__()
        self.stop_event = threading.Event()
        self.setDaemon(True)
        self.alive = True
 
        # ---- Setting for log ----
        self.log_path = './MCProtocol/Log/' 
        self.logger = getLogger(__name__)
        log.set_log_config(self.logger, self.log_path, __name__+'.log')
 
        # ---- Setting for MC Protocol ----
        self.target_plc_ip = '192.168.1.2'
        self.port = 1025 # Need to set config on PLC side
 
        # Prepare Mitsubishi Q connection
        self.mc = pymcprotocol.Type3E()
 
        return
 
 
    def __del__(self):
        print('End')
 
 
    def stop(self):
        self.stop_event.set()
 
 
    def run(self):
 
        # ---- Start connection ----
        ascii_flag = True
        self.__connect_plc(ascii_flag)
 
        # ---- Initialize local flag ----
        flag_shoot_done = False
        
 
        while not self.stop_event.is_set():
 
            try:
                # Initialize array for PLC bits
                bitunits_X = []
                bitunits_Y = []
 
                # Read the number of bitunits from headdevice 
                bitunits_X = self.mc.batchread_bitunits(headdevice='X0', readsize=3)
                bitunits_Y = self.mc.batchread_bitunits(headdevice='Y0', readsize=3) # ex. bitunits_Y array [Y0, Y1, Y2]
 
                # Y0 OFF
                if bitunits_Y[0] == 0:
 
                    if flag_shoot_done:
                        flag_shoot_done = False
 
                        self.logger.info('Y0 OFF')
                        bitunits_X[0] = 0
                        self.mc.batchwrite_bitunits(headdevice='X0', values=bitunits_X)
                        self.logger.info(f'Write {bitunits_X}')
 
                # Y0 ON
                if bitunits_Y[0] == 1:
 
                    if not flag_shoot_done:
                        flag_shoot_done = True
 
                        self.logger.info(f'Y0 ON')
                        bitunits_X[0] = 1
                        self.mc.batchwrite_bitunits(headdevice='X0', values=bitunits_X)
                        self.logger.info(f'Write {bitunits_X}')
                        
 
            except Exception as e:
                self.logger.exception(e)
                self.__disconnect_plc()
 
                # Delay for restarting
                time.sleep(5)
 
                self.logger.info('Re-connect')
                self.__connect_plc(ascii_flag)
 
 
        self.__disconnect_plc()
 
 
    def __connect_plc(self, ascii_flag):
        
        try:
            if ascii_flag:
                self.mc.setaccessopt(commtype='ascii')
            
            self.mc.connect(self.target_plc_ip, self.port)
            self.logger.info(f'Connected: IP {self.target_plc_ip}, Port {self.port} ')
 
        except Exception as e:
            self.logger.exception(e)
 
    def __disconnect_plc(self):
        try:
            self.mc.close()
            self.logger.info(f'Close connection')
 
        except Exception as e:
            self.logger.exception(e)
 
 
if __name__ == "__main__":
 
    ith = InspectionThread()
 
    ith.start()
    time.sleep(20)
    ith.stop()
    ith.join()

次に「log.py」。こちらはファイル名を変える場合は「mcp_test.py」のimport logの部分もそれに合わせて変えてください。

 #!/usr/bin/env python
# -*- coding: utf-8 -*-
from logging import StreamHandler, DEBUG, INFO, Formatter
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
import os
 
def set_log_config(logger, dir, filename):
 
    # ---- Prepare log folder ----
    os.makedirs(dir, exist_ok=True) 
    filepath = dir + filename
 
    # ---- handler1: For terminal ----
    handler1 = StreamHandler()
    handler1.setLevel(DEBUG)
    handler1.setFormatter(Formatter("[%(asctime)s][%(levelname)8s][%(name)s][%(funcName)s] %(message)s"))
 
    # ---- handler2: For log file ----
    #handler2 = TimedRotatingFileHandler(filename = filepath, when='midnight', backupCount=30, encoding='utf-8') # If you want to rotate file by time, use this
    handler2 = RotatingFileHandler(filename = filepath, maxBytes = 1048576, backupCount = 10, encoding='utf-8', ) # 1MB, keep 10 files
    handler2.setLevel(INFO)
    handler2.setFormatter(Formatter("[%(asctime)s][%(levelname)8s][%(name)s][%(funcName)s] %(message)s"))
 
    # ---- Set log config ----
    logger.setLevel(DEBUG)
    logger.addHandler(handler1)
    logger.addHandler(handler2)
    logger.propagate = False

テスト結果

設定1と2はどちらでもいけました。送信受信の設定は関係ないようですね。

設定1：TCP、Unpassive、受信、手順あり

エラーなく読み書きOK。PLC側でもX0がON/OFFしているのが確認できます。

設定2：TCP、Unpassive、送信、手順あり

こちらもエラーなく読み書きOK。PLC側でもX0がON/OFFしているのが確認できます。固定バッファは送信だろうが受信だろうが関係ないようですね。

設定3：TCP、Unpassive、受信、無手順

接続はできましたが、読み出しでエラー。

設定4：TCP、Unpassive、受信、事前定義プロトコル

接続はできましたが、読み出しでエラー。

つまり設定1か2を使えってことですね。

Pythonコードの解説

pymcprotocolの解説はこちらに記載しています。

ラズパイPythonと三菱PLCをネットワーク経由で連携させる方法

ログの解説はこちら。

結局どうする？ラクして無難にPython loggingを使うためのサンプルコードと使用例

流れ

直接コードを実行するとこの部分が走ります。

 if __name__ == "__main__":
 
    ith = InspectionThread()
 
    ith.start()
    time.sleep(20)
    ith.stop()
    ith.join()

InspectionThreadクラスをインスタンス化してスレッドを起動、20秒待ってスレッド停止、スレッドが終わるまでプログラムが終了しないよう待つ、という流れです。

今回はシングルスレッドですが、作り自体はThreadを継承してオーバーライドする、いわゆるPythonのスレッディング手法のひとつです。run内にメインプログラムを記述し、実行時はstart()。GUIなどから起動するためにスレッドにしています。

複数のインスタンスを持ちたい場合は、getLogger()に与える名前を変えれるようにするかクラスの外に出すとよいです。

最後に

いかがでしたか。これを使えば、ネットワーク経由でPLCからトリガーを取り込んでLiDARで画像検査、なんてのができますね。インテルのLiDAR、RealSenseをラズパイOSのPythonから起動するための記事はこちら。

RealSense L515をラズパイ4 Raspberry Pi OSのPythonから起動する最速手順＆サンプルコード

ラズパイのCPUクロックをPythonでPID制御しCPU温度コントロール

ラズパイPythonと三菱PLCのネットワーク連携【イーサネットユニット編】

この記事でわかること

前提条件

通信設定

イーサネットユニットパラメータ設定

ラズパイのIP設定

pymcprotocolのインストール

pymcprotocolを使った通信テスト

テスト準備

ラダー記述

Pythonコード記述

テスト結果

設定1：TCP、Unpassive、受信、手順あり

設定2：TCP、Unpassive、送信、手順あり

設定3：TCP、Unpassive、受信、無手順

設定4：TCP、Unpassive、受信、事前定義プロトコル

Pythonコードの解説

流れ

最後に

0 件のコメント:

コメントを投稿

人気の投稿

フォロワー

Labels

Pages

Search this blog

Author

	#!/usr/bin/env python
	# -- coding: utf-8 --
	import pymcprotocol
	import threading
	import log # 自作モジュール
	from logging import getLogger
	import time


	class InspectionThread(threading.Thread):

	def __init__(self):

	# ---- Setting for threading
	super(InspectionThread, self).__init__()
	self.stop_event = threading.Event()
	self.setDaemon(True)
	self.alive = True

	# ---- Setting for log ----
	self.log_path = './MCProtocol/Log/'
	self.logger = getLogger(__name__)
	log.set_log_config(self.logger, self.log_path, __name__+'.log')

	# ---- Setting for MC Protocol ----
	self.target_plc_ip = '192.168.1.2'
	self.port = 1025 # Need to set config on PLC side

	# Prepare Mitsubishi Q connection
	self.mc = pymcprotocol.Type3E()

	return


	def __del__(self):
	print('End')


	def stop(self):
	self.stop_event.set()


	def run(self):

	# ---- Start connection ----
	ascii_flag = True
	self.__connect_plc(ascii_flag)

	# ---- Initialize local flag ----
	flag_shoot_done = False


	while not self.stop_event.is_set():

	try:
	# Initialize array for PLC bits
	bitunits_X = []
	bitunits_Y = []

	# Read the number of bitunits from headdevice
	bitunits_X = self.mc.batchread_bitunits(headdevice='X0', readsize=3)
	bitunits_Y = self.mc.batchread_bitunits(headdevice='Y0', readsize=3) # ex. bitunits_Y array [Y0, Y1, Y2]

	# Y0 OFF
	if bitunits_Y[0] == 0:

	if flag_shoot_done:
	flag_shoot_done = False

	self.logger.info('Y0 OFF')
	bitunits_X[0] = 0
	self.mc.batchwrite_bitunits(headdevice='X0', values=bitunits_X)
	self.logger.info(f'Write {bitunits_X}')

	# Y0 ON
	if bitunits_Y[0] == 1:

	if not flag_shoot_done:
	flag_shoot_done = True

	self.logger.info(f'Y0 ON')
	bitunits_X[0] = 1
	self.mc.batchwrite_bitunits(headdevice='X0', values=bitunits_X)
	self.logger.info(f'Write {bitunits_X}')


	except Exception as e:
	self.logger.exception(e)
	self.__disconnect_plc()

	# Delay for restarting
	time.sleep(5)

	self.logger.info('Re-connect')
	self.__connect_plc(ascii_flag)


	self.__disconnect_plc()


	def __connect_plc(self, ascii_flag):

	try:
	if ascii_flag:
	self.mc.setaccessopt(commtype='ascii')

	self.mc.connect(self.target_plc_ip, self.port)
	self.logger.info(f'Connected: IP {self.target_plc_ip}, Port {self.port} ')

	except Exception as e:
	self.logger.exception(e)

	def __disconnect_plc(self):
	try:
	self.mc.close()
	self.logger.info(f'Close connection')

	except Exception as e:
	self.logger.exception(e)


	if __name__ == "__main__":

	ith = InspectionThread()

	ith.start()
	time.sleep(20)
	ith.stop()
	ith.join()

	#!/usr/bin/env python
	# -- coding: utf-8 --
	from logging import StreamHandler, DEBUG, INFO, Formatter
	from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
	import os

	def set_log_config(logger, dir, filename):

	# ---- Prepare log folder ----
	os.makedirs(dir, exist_ok=True)
	filepath = dir + filename

	# ---- handler1: For terminal ----
	handler1 = StreamHandler()
	handler1.setLevel(DEBUG)
	handler1.setFormatter(Formatter("[%(asctime)s][%(levelname)8s][%(name)s][%(funcName)s] %(message)s"))

	# ---- handler2: For log file ----
	#handler2 = TimedRotatingFileHandler(filename = filepath, when='midnight', backupCount=30, encoding='utf-8') # If you want to rotate file by time, use this
	handler2 = RotatingFileHandler(filename = filepath, maxBytes = 1048576, backupCount = 10, encoding='utf-8', ) # 1MB, keep 10 files
	handler2.setLevel(INFO)
	handler2.setFormatter(Formatter("[%(asctime)s][%(levelname)8s][%(name)s][%(funcName)s] %(message)s"))

	# ---- Set log config ----
	logger.setLevel(DEBUG)
	logger.addHandler(handler1)
	logger.addHandler(handler2)
	logger.propagate = False