pengqi/gateway_plc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

将读写功能修改为read_genetic,write_genetic并且添加.exe文件

Browse Source
This commit is contained in:
pengqi
2025-08-20 11:48:07 +08:00
parent bc0074824d
commit 83b374ffea
1177 changed files with 615 additions and 138357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
gateway/cache_manager.py
View File

@ -2,14 +2,15 @@ import threading
import time
import logging
from snap7.util import *
import struct
class CacheManager:
"""PLC数据缓存管理器"""
def __init__(self, config, plc_manager, app=None):
"""
初始化缓存管理器
Args:
config: 配置对象
plc_manager: PLC管理器实例
@ -28,7 +29,7 @@ class CacheManager:
self.plc_connection_status = {} # PLC连接状态
self.logger = logging.getLogger("CacheManager")
self.init_cache()
def init_cache(self):
"""初始化缓存结构"""
for plc in self.config["plcs"]:
@ -37,7 +38,7 @@ class CacheManager:
self.last_update[plc_name] = {}
self.plc_last_connected[plc_name] = 0 # 初始化为0（未连接）
self.plc_connection_status[plc_name] = "never_connected"
for area in plc["areas"]:
name = area["name"]
# 确保初始状态为断开
@ -51,7 +52,7 @@ class CacheManager:
"status": "disconnected" # 初始状态为断开
}
self.last_update[plc_name][name] = 0
def refresh_cache(self):
"""后台线程：定期刷新缓存"""
while self.running:
@ -62,10 +63,10 @@ class CacheManager:
plc_name = plc["name"]
refresh_interval = plc.get("refresh_interval", 0.5)
client = self.plc_manager.get_plc(plc_name)
# 检查PLC连接状态
plc_connected = client.connected
# 更新PLC连接状态
with self.lock:
if plc_connected:
@ -76,14 +77,14 @@ class CacheManager:
self.plc_connection_status[plc_name] = "never_connected"
else:
self.plc_connection_status[plc_name] = "disconnected"
# 刷新所有可读区域
for area in plc["areas"]:
if area["type"] in ["read", "read_write"]:
name = area["name"]
try:
data = client.read_db(area["db_number"], area["offset"], area["size"])
# 更新区域状态基于PLC连接状态和读取结果
with self.lock:
if plc_connected and data and len(data) == area["size"]:
@ -106,27 +107,23 @@ class CacheManager:
#计算需要睡眠的时间，确保总等于refresh_time
sleep_time = max(0, refresh_interval - execution_time)
time.sleep(sleep_time)
print(f"plc_name: {plc_name},"
f"Cache refresh completed.Execution time: {execution_time:.3f}s,"
f"Sleep time: {sleep_time:.3f}s,"
f"Total interval: {execution_time + sleep_time:.3f}s")
# 记录实际刷新间隔
self.logger.debug(f"plc_name: {plc_name},"
f"Cache refresh completed.Execution time: {execution_time:.3f}s,"
f"Sleep time: {sleep_time:.3f}s,"
f"Total interval: {execution_time + sleep_time:.3f}s")
time.sleep(self.refresh_interval)
except Exception as e:
self.logger.error(f"Error in refresh_cache: {e}")
time.sleep(self.refresh_interval)
def start(self):
"""启动缓存刷新线程"""
if self.running:
return
self.running = True
self.thread = threading.Thread(
target=self.refresh_cache,
@ -135,12 +132,12 @@ class CacheManager:
)
self.thread.start()
self.logger.info("Cache manager started")
def stop(self):
"""停止缓存刷新线程"""
if not self.running:
return
self.running = False
if self.thread:
# 等待线程结束，但设置超时防止卡死
@ -149,17 +146,17 @@ class CacheManager:
self.logger.warning("Cache refresh thread did not terminate gracefully")
self.thread = None
self.logger.info("Cache manager stopped")
def get_plc_connection_status(self, plc_name):
"""获取PLC连接状态"""
with self.lock:
return self.plc_connection_status.get(plc_name, "unknown")
def get_last_update_time(self, plc_name, area_name):
"""获取区域数据最后更新时间"""
with self.lock:
return self.last_update.get(plc_name, {}).get(area_name, 0)
def get_summary(self):
"""获取缓存摘要信息"""
summary = {}
@ -169,14 +166,14 @@ class CacheManager:
for area_name, area in areas.items():
last_update = self.last_update[plc_name][area_name]
plc_status = self.plc_connection_status.get(plc_name, "unknown")
# 区域状态应与PLC连接状态一致，除非有有效数据
area_status = area["status"]
if plc_status == "never_connected":
area_status = "never_connected"
elif plc_status == "disconnected" and self.last_update[plc_name][area_name] == 0:
area_status = "disconnected"
summary[plc_name][area_name] = {
"status": area_status,
"plc_connection_status": plc_status,
@ -185,24 +182,24 @@ class CacheManager:
"type": area["type"]
}
return summary
def get_area_status(self, plc_name, area_name):
"""获取区域状态"""
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
if not area:
return {"status": "not_found", "message": "PLC or area not found"}
plc_status = self.plc_connection_status.get(plc_name, "unknown")
last_update = self.last_update.get(plc_name, {}).get(area_name, 0)
# 区域状态应与PLC连接状态一致，除非有有效数据
area_status = area["status"]
if plc_status == "never_connected":
area_status = "never_connected"
elif plc_status == "disconnected" and last_update == 0:
area_status = "disconnected"
return {
"status": area_status,
"plc_connection_status": plc_status,
@ -211,24 +208,25 @@ class CacheManager:
"size": area["size"],
"type": area["type"]
}
def read_area(self, plc_name, area_name, offset, length):
"""单个区域读取"""
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
print("read area :",area)
if not area:
return None, "Area not found", "unknown", 0
if offset + length > area["size"]:
return None, "Offset out of bounds", "unknown", 0
client = self.plc_manager.get_plc(plc_name)
plc_status = self.plc_connection_status.get(plc_name, "unknown")
# 如果PLC未连接，直接返回错误
if plc_status != "connected":
return None, f"PLC not connected (status: {plc_status})", plc_status, 0
try:
data = client.read_db(area["db_number"], area["offset"] + offset, length)
# 验证数据有效性
@ -239,7 +237,7 @@ class CacheManager:
update_time = time.time()
self.last_update[plc_name][area_name] = update_time
area["status"] = "connected"
return data, None, plc_status, update_time
else:
area["status"] = plc_status
@ -287,29 +285,29 @@ class CacheManager:
area["status"] = plc_status
self.logger.error(f"Read failed for {plc_name}/{area_name}: {e}")
return None, f"Read failed: {str(e)}", plc_status, 0
def write_area(self, plc_name, area_name, offset, data):
"""单个区域写入"""
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
if not area:
return False, "Area not found", "unknown", 0
if area["type"] not in ["write", "read_write"]:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return False, "Area is read-only", plc_status, 0
if offset + len(data) > area["size"]:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return False, "Offset out of bounds", plc_status, 0
client = self.plc_manager.get_plc(plc_name)
plc_status = self.plc_connection_status.get(plc_name, "unknown")
# 如果PLC未连接，直接返回错误
if plc_status != "connected":
return False, f"PLC not connected (status: {plc_status})", plc_status, 0
try:
success = client.write_db(area["db_number"], area["offset"] + offset, data)
if success:
@ -319,7 +317,7 @@ class CacheManager:
update_time = time.time()
self.last_update[plc_name][area_name] = update_time
area["status"] = "connected (last write)"
return True, None, plc_status, update_time
else:
area["status"] = plc_status
@ -464,7 +462,7 @@ class CacheManager:
area["status"] = plc_status
self.logger.error(f"Write failed for {plc_name}/{area_name}: {e}")
return False, f"Write failed: {str(e)}", plc_status, 0
def batch_read(self, requests):
"""批量读取"""
results = []
@ -633,26 +631,234 @@ class CacheManager:
"length": len(data)
})
return results
def read_generic(self, plc_name, area_name, offset, data_type, count=1):
"""通用读取接口"""
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
print("area:",area)
if not area:
return None, "Area not found", "unknown", 0
if area["type"] not in ["read", "read_write"]:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return None, "Area is read-only", plc_status, 0
# 计算实际DB偏移
db_offset = area["offset"] + offset
# 确保在区域内
if data_type == 'bool':
required_size = (offset + count + 7) // 8
elif data_type in ['int', 'word']:
required_size = 2 * count
elif data_type in ['dint', 'dword', 'real']:
required_size = 4 * count
else: # byte
required_size = count
if db_offset + required_size > area["size"] or db_offset < 0:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return None, "Offset out of bounds", plc_status, 0
client = self.plc_manager.get_plc(plc_name)
plc_status = self.plc_connection_status.get(plc_name, "unknown")
# 如果PLC未连接，直接返回错误
if plc_status != "connected":
return None, f"PLC not connected (status: {plc_status})", plc_status, 0
try:
# 使用Snap7Client的read_generic方法
result = client.read_generic(area["db_number"], db_offset, data_type, count)
if result is None:
return None, "Read failed", plc_status, 0
# 对于bool类型，需要特殊处理缓存
if data_type == 'bool':
for i in range(count):
byte_offset = offset // 8 + i // 8
bit_offset = (offset % 8) + (i % 8)
if bit_offset >= 8:
byte_offset += 1
bit_offset -= 8
# 读取当前字节值
current_byte = area["data"][byte_offset]
if result[i]:
# 设置位为1
new_byte = current_byte | (1 << bit_offset)
else:
# 设置位为0
new_byte = current_byte & ~(1 << bit_offset)
area["data"][byte_offset] = new_byte
else:
# 对于其他类型，直接更新缓存
if not isinstance(result, list):
result = [result]
if data_type == 'byte':
item_size = 1
elif data_type in ['int', 'word']:
item_size = 2
else: # dint, dword, real
item_size = 4
for i, val in enumerate(result):
item_offset = offset + i * item_size
if data_type == 'byte':
area["data"][item_offset] = val & 0xFF
elif data_type in ['int', 'word']:
# 2字节数据
packed = struct.pack(">h" if data_type == "int" else ">H", val)
for j in range(2):
area["data"][item_offset + j] = packed[j]
elif data_type in ['dint', 'dword', 'real']:
# 4字节数据
packed = struct.pack(
">l" if data_type == "dint" else
">I" if data_type == "dword" else
">f",
val
)
for j in range(4):
area["data"][item_offset + j] = packed[j]
update_time = time.time()
self.last_update[plc_name][area_name] = update_time
area["status"] = "connected"
return result, None, plc_status, update_time
except Exception as e:
area["status"] = plc_status
self.logger.error(f"Read failed for {plc_name}/{area_name}: {e}")
return None, f"Read failed: {str(e)}", plc_status, 0
def write_generic(self, plc_name, area_name, offset, data_type, value):
"""通用写入接口"""
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
if not area:
return False, "Area not found", "unknown", 0
if area["type"] not in ["write", "read_write"]:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return False, "Area is read-only", plc_status, 0
# 计算实际DB偏移
db_offset = area["offset"] + offset
# 确保在区域内
if data_type == 'bool':
# 确定存储这些布尔值需要多少字节。
required_size = (offset + (len(value) if isinstance(value, list) else 1) + 7) // 8
elif data_type in ['int', 'word']:
required_size = 2 * (len(value) if isinstance(value, list) else 1)
elif data_type in ['dint', 'dword', 'real']:
required_size = 4 * (len(value) if isinstance(value, list) else 1)
else: # byte
required_size = len(value) if isinstance(value, list) else 1
if db_offset + required_size > area["size"] or db_offset < 0:
plc_status = self.plc_connection_status.get(plc_name, "unknown")
return False, "Offset out of bounds", plc_status, 0
client = self.plc_manager.get_plc(plc_name)
plc_status = self.plc_connection_status.get(plc_name, "unknown")
# 如果PLC未连接，直接返回错误
if plc_status != "connected":
return False, f"PLC not connected (status: {plc_status})", plc_status, 0
try:
# 使用Snap7Client的write_generic方法
success = client.write_generic(area["db_number"], db_offset, data_type, value)
if success:
# 根据数据类型更新缓存
if data_type == 'bool':
# 处理bool写入
if not isinstance(value, list):
value = [value]
for i, val in enumerate(value):
byte_offset = offset // 8 + i // 8
bit_offset = (offset % 8) + (i % 8)
if bit_offset >= 8:
byte_offset += 1
bit_offset -= 8
# 读取当前字节值
current_byte = area["data"][byte_offset]
if val:
# 设置位为1
new_byte = current_byte | (1 << bit_offset)
else:
# 设置位为0
new_byte = current_byte & ~(1 << bit_offset)
area["data"][byte_offset] = new_byte
elif data_type == 'byte':
# 处理byte写入
if not isinstance(value, list):
value = [value]
for i, val in enumerate(value):
area["data"][offset + i] = val & 0xFF
elif data_type in ['int', 'word']:
# 处理int/word写入
if not isinstance(value, list):
value = [value]
for i, val in enumerate(value):
# 2字节数据
packed = struct.pack(">h" if data_type == "int" else ">H", val)
for j in range(2):
area["data"][offset + i * 2 + j] = packed[j]
elif data_type in ['dint', 'dword', 'real']:
# 处理dint/dword/real写入
if not isinstance(value, list):
value = [value]
for i, val in enumerate(value):
# 4字节数据
packed = struct.pack(
">l" if data_type == "dint" else
">I" if data_type == "dword" else
">f",
val
)
for j in range(4):
area["data"][offset + i * 4 + j] = packed[j]
update_time = time.time()
self.last_update[plc_name][area_name] = update_time
area["status"] = "connected (last write)"
return True, None, plc_status, update_time
else:
area["status"] = plc_status
return False, "Write failed", plc_status, 0
except Exception as e:
area["status"] = plc_status
self.logger.error(f"Write failed for {plc_name}/{area_name}: {e}")
return False, f"Write failed: {str(e)}", plc_status, 0
def get_parsed_data(self, plc_name, area_name):
"""获取解析后的数据"""
from data_parser import parse_data
with self.lock:
area = self.cache.get(plc_name, {}).get(area_name)
if not area:
return {"error": "Area not found"}
plc_status = self.plc_connection_status.get(plc_name, "unknown")
last_update = self.last_update.get(plc_name, {}).get(area_name, 0)
# 区域状态应与PLC连接状态一致，除非有有效数据
area_status = area["status"]
if plc_status == "never_connected":
area_status = "never_connected"
elif plc_status == "disconnected" and last_update == 0:
area_status = "disconnected"
structure = area.get("structure", [])
if structure:
parsed = parse_data(area["data"], structure)