用Python采用Modbus-Tcp的方式读取PLC模块数据

news/2024/10/20 5:37:40/

使用计算器得到需要的寄存器地址

这里PLC地址是83,对应的程序16进制读取地址是53

实际上由于PLC地址从1开始，所以这里实际地址应该是52，因为计算机从0开始

在这里插入图片描述

使用网络调试助手生成报文

在这里插入图片描述

使用Python中的内置函数int()。以下是将人员卡号’b’3b44’'转换为十进制的示例代码：

card_number = '3b44'
decimal_number = int(card_number, 16)
print(decimal_number)

使用response[-4:]获取了响应数据的后4个字节作为value96。然后，通过struct.unpack(‘>f’, value96)[0]将4字节的二进制字符串解包为单精度浮点数，并将其打印出来。

#实时电量
request = bytes.fromhex("00 20 00 00 00 06 01 03 00 5F 00 02 ")
client_socket.send(request)
response = client_socket.recv(1024)value96 = response[-4:]
value96 = struct.unpack('>f', value96)[0]
value96=value96*10.00
value96=round(value96,2)
print("实时电量 单精度浮点数: {:.2f}".format(value96))

value40 是一个包含两个字节的字节串，即 b’\x00\x00’,将其转换为二进制，并保留8位。

以下是将字节串转换为二进制并保留8位的示例代码：

value40 = b'\x00\x00'
binary_value = bin(int.from_bytes(value40, byteorder='big'))[2:].zfill(8)
print("状态:", binary_value)

DEMO代码

#!D:/workplace/python
# -*- coding: utf-8 -*-
# @File  : main0523_04.py
# @Author:Romulushe
# @Time    : 2023/5/23 10:38
# @Software: PyCharm
# @Use: PyCharm
import socket
import struct
import time
import binasciiinterval = 5
ip_address = ''#根据实际情况自定义
port_number = 502
polling_interval = float(interval)def print_binary_value(value, name):binary_value = bin(int.from_bytes(value, byteorder='big'))[2:].zfill(8)[::-1]print(f"{name}: {binary_value}")with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:try:client_socket.settimeout(3)client_socket.connect((ip_address, port_number))while True:try:##电压request = bytes.fromhex("00 22 00 00 00 06 01 03 00 53 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value84 = response[-4:]value84= struct.unpack('>f', value84)[0]# value84 = value84 * 10.00value84 = round(value84, 2)print("实时电压 单精度浮点数: {:.2f}".format(value84))##电流request = bytes.fromhex("00 24 00 00 00 06 01 03 00 4F 00 02  ")client_socket.send(request)response = client_socket.recv(1024)value80 = response[-4:]value80 = struct.unpack('>f', value80)[0]# value84 = value80 * 10.00value80 = round(value80, 2)print("实时电流 单精度浮点数: {:.2f}".format(value80))##实时温度1request = bytes.fromhex("00 2E 00 00 00 06 01 03 00 57 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value88 = response[-4:]value88 = struct.unpack('>f', value88)[0]# value88 = value88 * 10.00value88 = round(value88, 2)print("实时温度1 单精度浮点数: {:.2f}".format(value88))#实时温度2request = bytes.fromhex("00 2C 00 00 00 06 01 03 00 5B 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value92 = response[-4:]value92 = struct.unpack('>f', value92)[0]# value92 = value92 * 10.00value92 = round(value92, 2)print("实时温度2 单精度浮点数: {:.2f}".format(value92))##车速request = bytes.fromhex("00 31 00 00 00 06 01 03 00 3B 00 01 ")client_socket.send(request)response = client_socket.recv(1024)speed= binascii.hexlify(response)[-4:]speed = int(speed, 16)print("车速:", speed)# #实时电量request = bytes.fromhex("00 20 00 00 00 06 01 03 00 5F 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value96 = response[-4:]value96 = struct.unpack('>f', value96)[0]value96=value96*10.00value96=round(value96,2)print("实时电量 单精度浮点数: {:.2f}".format(value96))#充放电状态#超速报警#低电量报警#温度过高1#温度过高2#00 18 00 00 00 06 01 03 00 28 00 01request = bytes.fromhex("00 33 00 00 00 06 01 03 00 28 00 01 ")client_socket.send(request)response = client_socket.recv(1024)value40 = response[-2:]value40= bin(int.from_bytes(value40, byteorder='big'))[2:].zfill(8)[::-1]print("充放电状态:",value40[4])print("超速报警:", value40[0])print("低电量报警:", value40[2])print("温度1过高:", value40[3])print("温度2过高:", value40[7])#车牌号#request = bytes.fromhex("00 00 00 00 00 06 01 03 00 1E 00 02 ")  # 车牌号client_socket.send(request)response = client_socket.recv(1024)car_num= binascii.hexlify(response)[-8:-4]car_num = int(car_num, 16)print("车牌号:", car_num)#人员卡号request = bytes.fromhex("00 00 00 00 00 06 01 03 00 42 00 02 ")  # 人员卡号client_socket.send(request)response = client_socket.recv(1024)card = binascii.hexlify(response)[-8:-4]card = int(card, 16)print("人员卡号:", card)except socket.timeout:print('TIMEOUT ERROR: 服务器未及时响应')except Exception as e:print('CONNECT ERROR:', e)

DEMO结果

E:\software\python\python.exe E:/projects/Forklift/t2.py实时电压 单精度浮点数: 48.85实时电流 单精度浮点数: 0.26实时温度1 单精度浮点数: 31.10实时温度2 单精度浮点数: 30.85车速: 0实时电量 单精度浮点数: 68.52充放电状态: 0超速报警: 0低电量报警: 0温度1过高: 0温度2过高: 0车牌号: 15172人员卡号: 10763

其他：地址表

在这里插入图片描述

附带数据库&log存储的代码：

#!D:/workplace/python
# -*- coding: utf-8 -*-
# @File  : main0523_04.py
# @Author:Romulushe
# @Time    : 2023/5/23 10:38
# @Software: PyCharm
# @Use: PyCharm
import socket
import struct
import time
import binascii
import pymysql
import os,sys,datetime,logginginterval = 5
ip_address = ''#根据实际情况自定义
port_number = ''#根据实际情况自定义
polling_interval = float(interval)base_path = os.path.dirname(os.path.realpath(sys.argv[0]))def get_log_path():return os.path.join(base_path, 'logs')def cleanup_logs():log_path = get_log_path()current_time = time.time()for file_name in os.listdir(log_path):file_path = os.path.join(log_path, file_name)if os.path.isfile(file_path):creation_time = os.path.getctime(file_path)if current_time - creation_time > (3 * 24 * 60 * 60):os.remove(file_path)def configure_logging():log_path = get_log_path()os.makedirs(log_path, exist_ok=True)log_filename = get_log_filename()log_file = os.path.join(log_path, log_filename)logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', filename=log_file)def get_log_filename():now = datetime.datetime.now()return now.strftime("%Y-%m-%d_%H-%M.log")def create_new_log():log_path = get_log_path()log_files = os.listdir(log_path)if len(log_files) >= 20:oldest_file = min(log_files)os.remove(os.path.join(log_path, oldest_file))log_filename = get_log_filename()log_filepath = os.path.join(log_path, log_filename)return log_filepathdef check_log_size(log_filepath):log_size = os.path.getsize(log_filepath)if log_size > 2 * 1024 * 1024:# 创建新的日志文件new_log_filepath = create_new_log()try:shutil.move(log_filepath, new_log_filepath)return new_log_filepathexcept PermissionError:insert_log(logger, f'{log_filepath} {PermissionError}', log_filepath)time.sleep(0.1)return log_filepathreturn log_filepathdef insert_log(logger, log_message, log_filepath):log_filepath = check_log_size(log_filepath)# 创建文件处理器file_handler = RotatingFileHandler(log_filepath, maxBytes=2 * 1024 * 1024, backupCount=1)file_handler.setLevel(logging.DEBUG)formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')file_handler.setFormatter(formatter)# 添加文件处理器到日志记录器logger.addHandler(file_handler)try:logger.debug(log_message)except PermissionError:insert_log(logger, f'{log_message} {PermissionError}', log_filepath)time.sleep(0.1)  # 延迟0.1秒# 移除文件处理器logger.removeHandler(file_handler)
# 连接数据库
def connect_database(host, port, user, password, db_name):try:conn = pymysql.connect(host=host, port=port, user=user, password=password, db=db_name)# print("成功连接到数据库")return connexcept pymysql.Error as e:insert_log(logger, f'DATABASE CONNECT FAILED', log_filepath)# print(f"数据库连接失败: {e}")# 插入数据
def insert_data(conn, types,table_name, create_time, location_no, parameter_desc, location, param_value):try:cursor = conn.cursor()sql = f"INSERT INTO {table_name} (TYPE, CREATE_TIME, LOCATION_NO, PARAMETER_DESC, LOCATION, PARAMVALUE) " \f"VALUES (%s, %s, %s, %s, %s, %s)"cursor.execute(sql, (types,create_time, location_no, parameter_desc, location, param_value))conn.commit()# print(f"数据插入成功: {create_time}, {parameter_desc}: {param_value}")cursor.close()except pymysql.Error as e:insert_log(logger, f'DATABASE INSERT DATA FAILED', log_filepath)# print(f"数据插入失败: {e}")with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:configure_logging()cleanup_logs()log_filepath = create_new_log()logger = logging.getLogger()logger.setLevel(logging.DEBUG)try:client_socket.settimeout(3)client_socket.connect((ip_address, port_number))# 连接数据库host = ''#根据实际情况自定义port = 3306user = ''#根据实际情况自定义password = ''#根据实际情况自定义db_name = ''#根据实际情况自定义table_name =''#根据实际情况自定义try:# 连接数据库conn = connect_database(host, port, user, password, db_name)except Exception as e:print("COON ERROR:",e)while True:try:types ='叉车'location ='F2堆02'##电压request = bytes.fromhex("00 22 00 00 00 06 01 03 00 53 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value84 = response[-4:]value84= struct.unpack('>f', value84)[0]# value84 = value84 * 10.00value84 = round(value84, 2)print("实时电压 单精度浮点数: {:.2f}".format(value84))location_no ='F01-1'parameter_desc ='实时电压'param_value=value84insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)##电流request = bytes.fromhex("00 24 00 00 00 06 01 03 00 4F 00 02  ")client_socket.send(request)response = client_socket.recv(1024)value80 = response[-4:]value80 = struct.unpack('>f', value80)[0]# value84 = value80 * 10.00value80 = round(value80, 2)print("实时电流 单精度浮点数: {:.2f}".format(value80))location_no = 'F01-2'parameter_desc = '实时电流'param_value=value80insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)##实时温度1request = bytes.fromhex("00 2E 00 00 00 06 01 03 00 57 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value88 = response[-4:]value88 = struct.unpack('>f', value88)[0]# value88 = value88 * 10.00value88 = round(value88, 2)print("实时温度1 单精度浮点数: {:.2f}".format(value88))location_no = 'F01-3'parameter_desc = '实时温度1'param_value = value88insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)#实时温度2request = bytes.fromhex("00 2C 00 00 00 06 01 03 00 5B 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value92 = response[-4:]value92 = struct.unpack('>f', value92)[0]# value92 = value92 * 10.00value92 = round(value92, 2)print("实时温度2 单精度浮点数: {:.2f}".format(value92))location_no = 'F01-4'parameter_desc = '实时温度2'param_value = value92insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)##车速request = bytes.fromhex("00 31 00 00 00 06 01 03 00 3B 00 01 ")client_socket.send(request)response = client_socket.recv(1024)speed= binascii.hexlify(response)[-4:]speed = int(speed, 16)print("车速:", speed)location_no = 'F01-5'parameter_desc = '车速'param_value = speedinsert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)# #实时电量request = bytes.fromhex("00 20 00 00 00 06 01 03 00 5F 00 02 ")client_socket.send(request)response = client_socket.recv(1024)value96 = response[-4:]value96 = struct.unpack('>f', value96)[0]value96=value96*10.00value96=round(value96,2)print("实时电量 单精度浮点数: {:.2f}".format(value96))location_no = 'F01-6'parameter_desc = '实时电量'param_value = value96insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)#充放电状态#超速报警#低电量报警#温度过高1#温度过高2#00 18 00 00 00 06 01 03 00 28 00 01request = bytes.fromhex("00 33 00 00 00 06 01 03 00 28 00 01 ")client_socket.send(request)response = client_socket.recv(1024)value40 = response[-2:]value40= bin(int.from_bytes(value40, byteorder='big'))[2:].zfill(8)[::-1]print("充放电状态:",value40[4])print("超速报警:", value40[0])print("低电量报警:", value40[2])print("温度1过高:", value40[3])print("温度2过高:", value40[7])location_no = 'F01-7'parameter_desc = '充放电状态'param_value = value40[4]insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)location_no = 'F01-8'parameter_desc = '超速报警'param_value = value40[0]insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)location_no = 'F01-9'parameter_desc = '低电量报警'param_value = value40[2]insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)location_no = 'F01-10'parameter_desc = '温度1过高'param_value = value40[3]insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)location_no = 'F01-11'parameter_desc = '温度2过高'param_value = value40[7]insert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)# 车牌号#request = bytes.fromhex("00 00 00 00 00 06 01 03 00 1E 00 02 ")  # 车牌号client_socket.send(request)response = client_socket.recv(1024)car_num = binascii.hexlify(response)[-8:-4]car_num = int(car_num, 16)print("车牌号:", car_num)location_no = 'F01-12'parameter_desc = '车牌号'param_value = car_numinsert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)# 人员卡号request = bytes.fromhex("00 00 00 00 00 06 01 03 00 42 00 02 ")  # 人员卡号client_socket.send(request)response = client_socket.recv(1024)card = binascii.hexlify(response)[-8:-4]card = int(card, 16)print("人员卡号:", card)location_no = 'F01-13'parameter_desc = '人员卡号'param_value = cardinsert_data(conn, types, table_name, create_time, location_no, parameter_desc, location, param_value)except socket.timeout:print('TIMEOUT ERROR: 服务器未及时响应')except Exception as e:print('CONNECT ERROR:', e)

http://www.ppmy.cn/news/874352.html

相关文章

微服务架构+创建微服务工程（商品/订单）

微服务架构+创建微服务工程（商品/订单）

目录 1.微服务架构 1.1.单体应用架构 1.2.微服务应用 1.3 微服务架构的优势 1.4.微服务架构的缺点(挑战) 1.5. SpringCloud与微服务关系 1.6.SpringBoot和SpringCloud关系 2. 创建微服务工程 2.1.数据库 2.2.搭建父工程 2.2 创建公共模块 2.3.商品系统 2.4.订单微…
阅读更多...
开源项目-太阳神三国杀-QT4编译教程

开源项目-太阳神三国杀-QT4编译教程

待更中 转载于:https://www.cnblogs.com/sinpener/p/10734692.html
阅读更多...
python分析《水浒传》小说，对文本数据内容清洗、分词、统计、词云可视化展示

python分析《水浒传》小说，对文本数据内容清洗、分词、统计、词云可视化展示

《水浒传》这个名字说起来估计看到这篇博文的朋友都不会陌生到哪里去，这是我们中国古代四大名著之一，非常的引人入胜，今天想基于《水浒传》的小说数据来做一点分析工作。 由于用到的整部小说数据过大无法一次发表，这里只将本文用到…
阅读更多...
[聊天实录]关于开源软件

[聊天实录]关于开源软件

*** 09:33:00 开源软件是软件免费提供，但是不是相关服务要收费呢？ 杨中科 09:34:07 “开源软件是软件免费提供”，错！开源软件只是说源代码开放，但是并没有说免费提供，比如RedHat就是开源软件，但…
阅读更多...
文学-赋：洛神赋（三国曹植赋作）

文学-赋：洛神赋（三国曹植赋作）

ylbtech-文学-赋：洛神赋（三国曹植赋作） 《洛神赋》是中国三国时期曹魏文学家曹植创作的辞赋名篇。此赋虚构了作者自己与洛神的邂逅相遇和彼此间的思慕爱恋，洛神形象美丽绝伦，人神之恋飘渺迷离，但由于人神道…
阅读更多...
【毕业纪念设计】同窗杀1---人物技能一览

【毕业纪念设计】同窗杀1---人物技能一览

【毕业纪念设计】同窗杀1---人物技能一览 本文由CSDN-蚍蜉撼青松【主页：http://blog.csdn.net/howeverpf】原创，谢绝转载！ 刘势力武将一览(十) 老刘 势力：刘 性别：男 体力：3 技能： 不苟&…
阅读更多...
xlua热更新学习

xlua热更新学习

http://www.jianshu.com/p/dc4de5612d9e xlua 代码 下载： https://github.com/Tencent/xLua/blob/master/Assets/XLua/Doc/faq.md# 腾讯解决方案： http://mp.weixin.qq.com/s/2bY7A6ihK9IMcA0bOFyB-Q 介绍的很好： http://blog.csdn.net/guofe…
阅读更多...
内网穿透工具frp简易教程

内网穿透工具frp简易教程

frp项目 如果你想把家里的电脑当做服务器用，做一个网站或者游戏服务器什么的，肯定会遇到一个问题：由于没有公网IP，而且有家里的路由器把关，导致其他地方的人完全无法连接到服务器。这时候就需要内网穿透和端口映射工具…
阅读更多...
最新文章

Copyright @ 2022~2023