【Python】FANUC机器人OPC UA通信并记录数据

2024-04-11 18:04
文章标签 python 数据 记录 通信 机器人 ua fanuc opc

本文主要是介绍【Python】FANUC机器人OPC UA通信并记录数据，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

目录

    • 引言
    • 机器人仿真
    • 环境准备
    • 代码实现
      • 1. 导入库
      • 2. 设置参数
      • 3. 日志配置
      • 4. OPC UA通信
      • 5. 备份旧CSV文件
      • 6. 主函数
    • 总结

引言

 OPC UA（Open Platform Communications Unified Architecture）是一种跨平台的、开放的数据交换标准，常用于工业自动化领域。Python因其易用性和丰富的库支持，成为实现OPC UA通信的不错选择。本文将介绍如何使用Python进行OPC UA通信，并实时记录从FANUC机器人获取的数据。

机器人仿真

 FANUC机器人可以使用官方软件RoboGuide进行机器人仿真，启动后默认OPC UA地址为127.0.0.1:4880/FANUC/NanoUaServer。
在这里插入图片描述

环境准备

  • Python 3.5+
  • opcua库：用于实现OPC UA通信
  • logging库：用于记录日志

安装opcua库：

pip install opcua

代码实现

1. 导入库

import csv
from datetime import datetime
import logging
import os
import shutil
import time
from typing import List
from opcua.common.node import Node
from opcua import Client, ua

2. 设置参数

SERVER_URL = "opc.tcp://127.0.0.1:4880/FANUC/NanoUaServer"
CSV_FILENAME = 'fanuc_opcua_data.csv'
FAUNC_LOG = 'fanuc.log'
LOG_DIR = 'log'
BACKUP_DIR = 'backup'

3. 日志配置

def getLogger(filename: str):if not os.path.exists(LOG_DIR):os.makedirs(LOG_DIR)logger = logging.Logger(filename[:-4].upper(), logging.INFO)formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s %(message)s")fh = logging.FileHandler(LOG_DIR + "/" + filename, encoding="utf-8", mode="a")fh.setFormatter(formatter)ch = logging.StreamHandler()ch.setFormatter(formatter)logger.addHandler(fh)logger.addHandler(ch)return loggerLOGGER = getLogger(FAUNC_LOG)

4. OPC UA通信

  • 连接到服务器
def connect_to_server(url):client = Client(url)client.connect()return client
  • 获取根节点和对象节点
def get_root_node(client: Client):return client.get_root_node()
def get_objects_node(client: Client):return client.get_objects_node()
  • 遍历所有子节点并返回变量节点的路径和数值
def get_variables(node: Node, path=""):variables = {}children: List[Node] = node.get_children()for child in children:try:name: ua.QualifiedName = child.get_browse_name()new_path = f"{path}/{name.Name}"if child.get_node_class() == ua.NodeClass.Variable:value = child.get_value()if isinstance(value, list):value = ','.join(str(x) for x in value)if isinstance(value, str):value = value.replace('\n', '\\n').replace(',', ' ')variables[new_path] = valueelse:variables.update(get_variables(child, new_path))except Exception as e:LOGGER.error(f"Error fetching variable: {new_path}, Error: {e}")return variables

5. 备份旧CSV文件

def backup_csv_file(filename):if not os.path.exists(BACKUP_DIR):os.makedirs(BACKUP_DIR)if os.path.exists(filename):modification_time = os.path.getmtime(filename)modification_time_str = datetime.fromtimestamp(modification_time).strftime('%Y%m%d%H%M%S')new_filename = f"{BACKUP_DIR}/{filename}_{modification_time_str}"try:shutil.move(filename, new_filename)LOGGER.info(f"文件已移动到 {new_filename}")except Exception as e:LOGGER.error(f"移动文件出错: {new_filename}, Error: {e})

6. 主函数

if __name__ == "__main__":try:client = connect_to_server(SERVER_URL)root_node = get_root_node(client)objects_node = get_objects_node(client)backup_csv_file(CSV_FILENAME)with open(CSV_FILENAME, mode='w', newline='') as csvfile:num = 0while True:variables = get_variables(objects_node)if num == 1:writer = csv.DictWriter(csvfile, fieldnames=variables.keys())writer.writeheader()writer.writerow(variables)csvfile.flush()num += 1LOGGER.info("数据记录:" + str(num))time.sleep(1)except KeyboardInterrupt:print("程序被用户中断")finally:client.disconnect()

记录数据预览：
在这里插入图片描述

总结

 本文介绍了如何使用Python进行OPC UA通信，并实时记录从FANUC机器人获取的数据。通过使用opcua库，我们可以轻松地连接到OPC UA

完整代码：

import csv
from datetime import datetime
import logging
import os
import shutil
import time
from typing import List
from opcua.common.node import Node
from opcua import Client, ua# OPC UA服务器的URL
SERVER_URL = "opc.tcp://127.0.0.1:4880/FANUC/NanoUaServer"
# CSV文件名
CSV_FILENAME = 'fanuc_opcua_data.csv'
# 日志文件
FAUNC_LOG = 'fanuc.log'
# 文件夹 
LOG_DIR = 'log'
BACKUP_DIR = 'backup'def getLogger(filename: str):if not os.path.exists(LOG_DIR):os.makedirs(LOG_DIR)logger = logging.Logger(filename[:-4].upper(), logging.INFO)formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s %(message)s")fh = logging.FileHandler(LOG_DIR + "/" + filename, encoding="utf-8", mode="a")fh.setFormatter(formatter)ch = logging.StreamHandler()ch.setFormatter(formatter)logger.addHandler(fh)logger.addHandler(ch)return loggerLOGGER = getLogger(FAUNC_LOG)
def connect_to_server(url):"""创建客户端实例并连接到服务端"""client = Client(url)client.connect()return clientdef get_root_node(client: Client):"""获取服务器命名空间中的根节点"""return client.get_root_node()def get_objects_node(client: Client):"""获取服务器的对象节点"""return client.get_objects_node()def get_variables(node: Node, path=""):"""遍历所有子节点并返回变量节点的路径和数值"""variables = {}children: List[Node] = node.get_children()for child in children:try:name: ua.QualifiedName = child.get_browse_name()new_path = f"{path}/{name.Name}"if child.get_node_class() == ua.NodeClass.Variable:value = child.get_value()if isinstance(value, list):value = ','.join(str(x) for x in value)if isinstance(value, str):value = value.replace('\n', '\\n').replace(',', ' ')variables[new_path] = valueelse:variables.update(get_variables(child, new_path))except Exception as e:LOGGER.error(f"Error fetching variable: {new_path}, Error: {e}")return variablesdef backup_csv_file(filename):"""如果CSV文件已存在则备份"""if not os.path.exists(BACKUP_DIR):os.makedirs(BACKUP_DIR)if os.path.exists(filename):modification_time = os.path.getmtime(filename)modification_time_str = datetime.fromtimestamp(modification_time).strftime('%Y%m%d%H%M%S')new_filename = f"{BACKUP_DIR}/{filename}_{modification_time_str}"try:shutil.move(filename, new_filename)LOGGER.info(f"文件已移动到 {new_filename}")except Exception as e:LOGGER.error(f"移动文件出错: {new_filename}, Error: {e}")if __name__ == "__main__":try:client = connect_to_server(SERVER_URL)root_node = get_root_node(client)objects_node = get_objects_node(client)backup_csv_file(CSV_FILENAME)with open(CSV_FILENAME, mode='w', newline='') as csvfile:num = 0while True:variables = get_variables(objects_node)if num == 1:writer = csv.DictWriter(csvfile, fieldnames=variables.keys())writer.writeheader()writer.writerow(variables)csvfile.flush()num += 1LOGGER.info("数据记录:" + str(num))time.sleep(1)except KeyboardInterrupt:print("程序被用户中断")finally:client.disconnect()

这篇关于【Python】FANUC机器人OPC UA通信并记录数据的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/894798。

相关文章

51单片机学习记录———定时器

51单片机学习记录———定时器

文章目录 前言一、定时器介绍二、STC89C52定时器资源三、定时器框图四、定时器模式五、定时器相关寄存器六、定时器练习 前言 一个学习嵌入式的小白~ 有问题评论区或私信指出~ 提示：以下是本篇文章正文内容，下面案例可供参考 一、定时器介绍 定时器介绍：51单片机的定时器属于单片机的内部资源，其电路的连接和运转均在单片机内部完成。 定时器作用： 1.用于计数系统，可
阅读更多...
Javascript高级程序设计(第四版)--学习记录之变量、内存

Javascript高级程序设计(第四版)--学习记录之变量、内存

原始值与引用值 原始值：简单的数据即基础数据类型，按值访问。 引用值：由多个值构成的对象即复杂数据类型，按引用访问。 动态属性 对于引用值而言，可以随时添加、修改和删除其属性和方法。 let person = new Object();person.name = 'Jason';person.age = 42;console.log(person.name,person.age);//'J
阅读更多...
vcpkg安装opencv中的特殊问题记录（无法找到opencv_corexd.dll）

vcpkg安装opencv中的特殊问题记录（无法找到opencv_corexd.dll）

我是按照网上的vcpkg安装opencv方法进行的（比如这篇：从0开始在visual studio上安装opencv（超详细，针对小白）），但是中间出现了一些别人没有遇到的问题，虽然原因没有找到，但是本人给出一些暂时的解决办法： 问题1： 我在安装库命令行使用的是 .\vcpkg.exe install opencv 我的电脑是x64，vcpkg在这条命令后默认下载的也是opencv2:x6
阅读更多...
Python 字符串占位

Python 字符串占位

在Python中，可以使用字符串的格式化方法来实现字符串的占位。常见的方法有百分号操作符 % 以及 str.format() 方法 百分号操作符 % name = "张三"age = 20message = "我叫%s，今年%d岁。" % (name, age)print(message) # 我叫张三，今年20岁。 str.format() 方法 name = "张三"age
阅读更多...
通信系统网络架构_2.广域网网络架构

通信系统网络架构_2.广域网网络架构

1.概述          通俗来讲，广域网是将分布于相比局域网络更广区域的计算机设备联接起来的网络。广域网由通信子网于资源子网组成。通信子网可以利用公用分组交换网、卫星通信网和无线分组交换网构建，将分布在不同地区的局域网或计算机系统互连起来，实现资源子网的共享。 2.网络组成          广域网属于多级网络，通常由骨干网、分布网、接入网组成。在网络规模较小时，可仅由骨干网和接入网组成
阅读更多...
【服务器运维】MySQL数据存储至数据盘

【服务器运维】MySQL数据存储至数据盘

查看磁盘及分区 [root@MySQL tmp]# fdisk -lDisk /dev/sda: 21.5 GB, 21474836480 bytes255 heads, 63 sectors/track, 2610 cylindersUnits = cylinders of 16065 * 512 = 8225280 bytesSector size (logical/physical)
阅读更多...
ROS话题通信流程自定义数据格式

ROS话题通信流程自定义数据格式

ROS话题通信流程自定义数据格式 需求流程实现步骤定义msg文件编辑配置文件编译 在 ROS 通信协议中，数据载体是一个较为重要组成部分，ROS 中通过 std_msgs 封装了一些原生的数据类型,比如:String、Int32、Int64、Char、Bool、Empty… 但是，这些数据一般只包含一个 data 字段，结构的单一意味着功能上的局限性，当传输一些复杂的数据，比如:
阅读更多...
记录AS混淆代码模板

记录AS混淆代码模板

开启混淆得先在build.gradle文件中把 minifyEnabled false改成true,以及shrinkResources true//去除无用的resource文件 这些是写在proguard-rules.pro文件内的 指定代码的压缩级别 -optimizationpasses 5 包明不混合大小写 -dontusemixedcaseclassnames 不去忽略非公共
阅读更多...
一道经典Python程序样例带你飞速掌握Python的字典和列表

一道经典Python程序样例带你飞速掌握Python的字典和列表

Python中的列表（list）和字典（dict）是两种常用的数据结构，它们在数据组织和存储方面有很大的不同。 列表（List） 列表是Python中的一种有序集合，可以随时添加和删除其中的元素。列表中的元素可以是任何数据类型，包括数字、字符串、其他列表等。列表使用方括号[]表示，元素之间用逗号,分隔。 定义和使用 # 定义一个列表 fruits = ['apple', 'banana
阅读更多...
Python应用开发——30天学习Streamlit Python包进行APP的构建（9）

Python应用开发——30天学习Streamlit Python包进行APP的构建（9）

st.area_chart 显示区域图。 这是围绕 st.altair_chart 的语法糖。主要区别在于该命令使用数据自身的列和指数来计算图表的 Altair 规格。因此，在许多 "只需绘制此图 "的情况下，该命令更易于使用，但可定制性较差。 如果 st.area_chart 无法正确猜测数据规格，请尝试使用 st.altair_chart 指定所需的图表。 Function signa
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2