製造業の現場では生産を継続するための保守が重要であり、そのための生産設備のデータ収集および不具合発生の要因特定が重要な課題となっています。ただし、多数の設備が稼働する工場では、設備ごとに異なるデータフォーマットや通信方式の違いにより、データの収集・管理が困難です。本記事では、KVシリーズを活用したPLC一括監視システムの構築について解説します。
🔍 こんな方におすすめ！✅ 工場のDX化を進めたいがどうすればいいかわからない
✅ 設備監視システムの導入を検討している
✅ 簡単かつ素早くデータ収集を実現したい
一元監視システムの構築に着手する場合、以下の課題がよく挙げられます。
キーエンスのKVシリーズは、これらの課題に対してメリットがあります。
KVシリーズを活用することで、既存のPLCを大きく変更せずにデータを収集できます。
.png)
本ブログでは、高度な分析を行うための外部PCとの連携方法について解説します。
以下のステップでシステム構築を進めます。
4.1 KV-8000のFTPサーバ設定
4.2 KV-8000のロギング/トレース設定
4.3 KV-8000のテストプログラム作成
4.4 KV-8000の動作チェック
4.5 Pythonの分析用プログラム作成
4.6 全体動作チェック
KV-STUDIOのユニットエディタからKV-8000の設定を行います。
.png)
.png)
ロギング/トレース設定を行います。
log0(ID0)で以下の項目を設定します。
.png)
.png)
.png)
.png)
.png)
.png)
次に、log0の計測を開始するためのプログラムを追加します。
.png)
.png)
.png)
テストプログラムを作成します。
スキャン周期ごとにDM00005を1ずつ加算し、100になったら0にリセットする動作を行います。
注意：
4.2ではDM00005を記録するための周期(ロギング周期)を1000ミリ秒毎に設定しています。DM00005はスキャン周期毎に計算されますが、ロギング周期が非常に粗いため、DM00005の細かい変化は記録できません。
WindowsPCとKV-8000のEthernet/IPポートをUSBで接続してください。また、SDカードをセットしてください。
.png)
4.1~4.3で設定したプログラムをRUNモードで転送してください。
.png)
WindowsPCのイーサネット設定をしてください。
WindowsPCはIPアドレス 192.168.1.200、サブネットマスク255.255.255.0を設定します。
.png)
WindowsPCとKV-8000をLANケーブルで接続してください。
接続確認を行います。WindowsPCでコマンドプロンプトを開き、ping 192.168.1.1を入力および実行してください。
[接続失敗]
.png)
[接続成功]
.png)
.png)
FTP通信を開始するコマンドを実行します。
WindowsPCでコマンドプロンプトを開き、ftp 192.168.1.1を入力および実行してください。
通信に成功するとユーザとパスワードが聞かれます。KV-8000に設定したFTP通信用のパスワードを入力してください。今回は両方ともKVと入力します。
.png)
[接続成功]
.png)
接続に成功したら、以下のコマンドを入力してlog0フォルダ移動します。
cd 0_CARD
cd log0
移動後、ls -lのコマンドを実行します。
log0フォルダ内のCSVファイル一覧が表示されます。
.png)
Pythonをインストールしてください。
記載するプログラムはPython3.11.0で動作確認を行いました。
https://www.python.org/downloads/release/python-3110/
以下のPythonライブラリを使用しています。インストールしてください。
[インストールコマンドの例]
pip install pandas
デスクトップ上にftptestフォルダを作成してください。
.png)
ftptestフォルダに空のPythonファイル(拡張子 .py)を4つ作成してください。
メモ帳でファイル名＋.py、ファイルの種類を変更して保存することで作成できます。
.png)
Pythonファイルの名称を設定してください。
4つのサンプルコードを、それぞれの対応するPythonファイルにコピーしてください。
# main.py
import time
from ftp_downloader import download_new_csv_files
from csv_analyzer import analyze_and_archive
if __name__ == "__main__":
while True:
print("[メイン] 処理ループ開始")
new_files = download_new_csv_files()
if new_files:
analyze_and_archive(new_files)
else:
print("[メイン] 新規ファイルなし")
time.sleep(10)
# kv_writer.py
import pymcprotocol
class SimpleKV8000Writer:
def __init__(self, ip: str, port: int):
self.pymc = pymcprotocol.Type4E()
self.pymc.setaccessopt(commtype="binary", timer_sec=5)
self.pymc.connect(ip, port)
def write_max_to_dm10(self, value: int):
print(f"[KV8000] Writing value {value} to D00010")
self.pymc.batchwrite_wordunits("D00010", [value])※DM00010はMCプロトコルにおいてD00010と表現されます
# ftp_downloader.py
from ftplib import FTP
import os
# KV8000のIPとFTPサーバ
FTP_HOST = '192.168.1.1'
FTP_USER = 'KV'
FTP_PASS = 'KV'
# FTPサーバのCSV保存先フォルダ
REMOTE_DIR = '/0_CARD/log0'
# ダウンロード済みCSVファイルの一次保存用フォルダ
LOCAL_DIR = './data/raw'
HISTORY_FILE = os.path.join(LOCAL_DIR, 'downloaded.txt')
def download_new_csv_files():
downloaded = set()
if os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE, 'r') as f:
downloaded = set(f.read().splitlines())
ftp = FTP(FTP_HOST)
ftp.login(FTP_USER, FTP_PASS)
ftp.cwd(REMOTE_DIR)
os.makedirs(LOCAL_DIR, exist_ok=True)
filenames = ftp.nlst()
new_files = []
for filename in filenames:
if filename.endswith('.csv') and filename not in downloaded:
local_filename = os.path.basename(filename)
local_path = os.path.join(LOCAL_DIR, local_filename)
with open(local_path, 'wb') as f:
ftp.retrbinary(f'RETR ' + filename, f.write)
print(f"[FTP] ダウンロード完了: {local_filename}")
downloaded.add(filename)
new_files.append(local_path)
ftp.quit()
# 重複ダウンロード防止機能
max_entries = 1000
all_downloaded = list(downloaded)
if len(all_downloaded) > max_entries:
all_downloaded = all_downloaded[-max_entries:]
else:
all_downloaded = sorted(all_downloaded)
with open(HISTORY_FILE, 'w') as f:
for name in all_downloaded:
f.write(name + '\n')
return new_files
# csv_analyzer.py
import os
import shutil
import pandas as pd
from datetime import datetime
from kv_writer import SimpleKV8000Writer
# KV8000のIPとMCプロトコルポート
KV8000_IP = "192.168.1.1"
KV8000_PORT = 5000
# ダウンロード済みCSVファイルの一次保存用フォルダ
RAW_DIR = './data/raw'
# 分析済みCSVファイルの保存用フォルダ
PROCESSED_DIR = './data/processed'
# 分析する列名の指定
VALUE_COLUMN = "DM00005"
def analyze_and_archive(csv_files):
os.makedirs(PROCESSED_DIR, exist_ok=True)
kv_writer = SimpleKV8000Writer(KV8000_IP, KV8000_PORT)
for filepath in csv_files:
max_val = None
try:
df = pd.read_csv(filepath, skiprows=1)
expected_cols = ["DATE", "TIME", VALUE_COLUMN]
if list(df.columns) != expected_cols:
raise ValueError(f"列名が一致しません: {df.columns}")
df["DATETIME"] = pd.to_datetime(df["DATE"] + " " + df["TIME"])
df[VALUE_COLUMN] = pd.to_numeric(df[VALUE_COLUMN], errors='coerce')
max_val = df[VALUE_COLUMN].max()
print(f"[分析] {os.path.basename(filepath)} 最大値:")
print(f" {VALUE_COLUMN}: {max_val}")
# KV8000のDM00010に最大値を書き込む
if pd.notna(max_val):
int_val = int(max_val)
kv_writer.write_max_to_dm10(int_val)
print(f"[送信] KV8000へ {int_val} をDM00010に書き込み")
except Exception as e:
print(f"[エラー] {filepath} の分析中に例外: {e}")
# 分析済みファイルの処理
dst = os.path.join(PROCESSED_DIR, os.path.basename(filepath))
shutil.move(filepath, dst)
print(f"[整理] 処理済へ移動: {dst}")
コマンドプロンプトを利用して、ftptestフォルダに移動してください。
cd ./Desktop/ftptest
.png)
main.pyを実行してください。
python main.py
プログラムが開始すると、以下のようなログが流れます。
.png)
KV-STUDIOの一括モニタからDM00010(図中DM10)の数値を確認します。送信ログと同時にDM00010の値が変わると成功です。
.png)
本記事では、高度な分析を行うための外部PCとの連携について解説しました。
クラウド連携やAI分析の導入も視野に入れて、スマートファクトリー実現の第一歩としてご活用ください。
🔍 さらに詳細な技術情報を知りたい方は、次回の記事をお楽しみに！
%20(1).jpg)
.png)