zhangwei
/
VMiddleware
generated from zhangwei/Train_Identify
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

更新V2.9 

1、优化开启报表时，所输入的股道编号大于配置文件中的股道数的对应报错内容
2、更改读取掐煤量的依据标签
This commit is contained in:
Mr.V 2024-03-04 15:46:11 +08:00
parent 0740405e43
commit db7eab598b
15 changed files with 2312 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Dockerfile Normal file
View File

@ -0,0 +1,15 @@
#基于的基础镜像
FROM python:3.7_arm
#FROM python:3.7
#代码添加到code文件夹
ADD ./app /code
#设置code文件夹是工作目录
WORKDIR /code
#安装支持
RUN pip install --default-timeout=1000 -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/
CMD ["python", "./app.py"]

13
README Normal file
View File

@ -0,0 +1,13 @@
# VMiddleware 是安家岭项目的PLC通讯中间件
## 环境
- python: 3.7
- Linux : Ubuntu 18.04
- PLC : AB
## 版本
### V2.8
- 更新多股道数据读取
### V2.9
- 更新所读股道超过配置文件中的“股道数”时，所产生的报错提示
- 更新读取掐煤量的时间参考点位

51
VMiddleware.spec Normal file
View File

@ -0,0 +1,51 @@
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
[],
exclude_binaries=True,
name='VMiddleware',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=['logo.png'],
)
coll = COLLECT(
exe,
a.binaries,
a.zipfiles,
a.datas,
strip=False,
upx=True,
upx_exclude=[],
name='VMiddleware',
)

334
app.py Normal file
View File

@ -0,0 +1,334 @@
# !flask/bin/python
import threading
from threading import Thread
from flask import Flask, jsonify, request
from flask_socketio import SocketIO, emit, send
# from gevent import pywsgi
# from threading import Thread
import os
import sys
import eventlet
import time
from utils.Tool import logger, config, traceback, PLC_Tool
import service.DetectService as detect
app = Flask(__name__)
app.config['SECRET_KEY'] = 'VMiddleware'
app.config['PROPAGATE_EXCEPTIONS'] = False
app.config['JSON_AS_ASCII'] = False
socketio = SocketIO(app)
eventlet.monkey_patch()
def responseMsg(code: int, msg: str, data):
if code in [500, 412]:
logger.error(msg)
elif code == 417:
logger.error("%s date:%s" % (msg, str(data)))
elif code == 200:
logger.info(msg)
return jsonify({'code': code, 'msg': msg, "data": data}), 200
def restart_program():
time.sleep(2)
logger.info("Restarting program...")
python = sys.executable
os.execl(python, python, *sys.argv)
@app.errorhandler(404)
def handle_not_found_error(error):
return responseMsg(404, '请查阅接口文档，使用正确地址', {}), 404
@app.route('/')
def hello(): # put application's code here
return responseMsg(404, '请查阅接口文档，使用正确地址', {}), 404
@app.route('/api/report_open', methods=['POST'])
def report_open():
try:
requestData = request.json
pound_no = requestData.get('pound_no', None)
total_number_carriage = requestData.get('total_number_carriage', 0)
pre_load = requestData.get('pre_load', [])
# if pound_no is None:
# return responseMsg(412, 'no pound_no', {})
# if not isinstance(pound_no, str):
# return responseMsg(412, "pound_no should be a string", {})
# if pound_no != "":
# return responseMsg(412, 'no pound_no', {})
# if isinstance(pound_no, str) and not pound_no.isdigit():
# return responseMsg(412, "pound_no should be a int and 0 < track <= 2", {})
# if not (0 < int(pound_no) <= 2):
# return responseMsg(412, "pound_no should be a int and 0 < track <= 2", {})
code, msg, results = detect.report_open(pound_no, total_number_carriage, pre_load)
return responseMsg(code, "开启PLC报表成功" if code == 200 else msg, results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/api/report_close', methods=['POST'])
def report_close():
try:
requestData = request.json
pound_no = requestData.get('pound_no', None)
total_number_carriage = requestData.get('total_number_carriage', 0)
code, msg, results = detect.report_close(pound_no, total_number_carriage)
return responseMsg(code, "关闭PLC报表成功" if code == 200 else msg, results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/api/read_config', methods=['GET'])
def read_config():
try:
return responseMsg(200, "读取成功", config.getConfig())
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/api/update_config', methods=['POST'])
def update_config():
try:
requestData = request.json
option = requestData.get('option', "")
value = requestData.get('value', "")
msg = config.updateConfig(option, value)
if msg != "":
return responseMsg(412, msg, {})
return responseMsg(200, "修改配置项 {%s: %s} 成功！" % (option, value), {})
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/api/restart_VMiddleware', methods=['GET'])
def restart_VMiddleware():
try:
t = threading.Thread(target=restart_program)
t.start()
return responseMsg(200, "程序重启...", {})
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/test/write/tags', methods=['POST'])
def write_tag():
plc = None
try:
requestData = request.json
plcIp = requestData.get("plc_ip", None)
plcSlot = requestData.get("slot", None)
plcTags = requestData.get("tags", None)
if plcIp is None:
return responseMsg(412, 'no plc_ip', {}) # 412 没有传输plc_ip，直接返回
if plcSlot is None:
return responseMsg(412, 'no slot', {}) # 412 没有传输slot，直接返回
if not isinstance(plcTags, dict):
return responseMsg(412, 'tags格式异常', {}) # 412 没有按格式传输tags，直接返回
tags = list(zip(plcTags.keys(), plcTags.values()))
if len(tags) == 0:
return responseMsg(412, 'no tags', {}) # 412 没有传输tags，直接返回
if isinstance(plcSlot, str) and not plcSlot.isdigit():
plcSlot = 0
plc = PLC_Tool(plcIp, plcSlot)
success, msg, results = plc.batch_write_tag(tags)
if not success:
return responseMsg(412, msg, {})
else:
if success > 1:
return responseMsg(417, msg, results)
else:
return responseMsg(200, "【测试功能】修改PLC标签值成功", results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
finally:
if plc is not None:
plc.close()
@app.route('/test/read/tags', methods=['GET'])
def get_tag():
plc = None
try:
plcIp = request.args.get('plcIp', None)
plcSlot = request.args.get('plcSlot', None)
tag = request.args.get('tags', None)
if plcIp is None:
return responseMsg(412, 'no plc_ip', {}) # 412 没有传输plc_ip，直接返回
if plcSlot is None:
return responseMsg(412, 'no slot', {}) # 412 没有传输slot，直接返回
if not tag.startswith('[') or not tag.endswith(']'):
return responseMsg(412, 'tags格式异常', {}) # 412 没有按格式传输tags，直接返回
tags = (tag[1:-1].replace(' ', '')).split(',')
if len(tags) < 1:
return responseMsg(412, 'no tags', {}) # 412 没有传输tags，直接返回
if isinstance(plcSlot, str) and not plcSlot.isdigit():
plcSlot = 0
plc = PLC_Tool(plcIp, plcSlot)
success, msg, results = plc.get_tags(tags)
if not success:
return responseMsg(412, msg, {})
else:
return responseMsg(200, "【测试功能】读取PLC标签值成功", results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
finally:
if plc is not None:
plc.close()
@app.route('/test/get_token', methods=['GET'])
def get_web_authorization():
try:
code, msg, results = detect.get_web_token()
return responseMsg(code, "【测试功能】获取token成功" if code == 200 else msg, results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/test/upload_pinch_coal', methods=['POST'])
def upload_pinch_coal():
try:
requestData = request.json
carriage_order = requestData.get("carriage_order", 0)
pinch_coal = requestData.get("pinch_coal", -1)
pound_no = requestData.get("pound_no", "")
code, msg, results = detect.upload_pinch_coal(carriage_order, pinch_coal, pound_no)
return responseMsg(code, "【测试功能】上传掐煤量成功" if code == 200 else msg, results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@app.route('/test/upload_gross_weight', methods=['POST'])
def upload_gross_weight():
try:
requestData = request.json
carriage_order = requestData.get("carriage_order", 0)
gross_weight = requestData.get("gross_weight", -1)
pound_no = requestData.get("pound_no", "")
code, msg, results = detect.upload_gross_weight(carriage_order, gross_weight, pound_no)
return responseMsg(code, "【测试功能】上传装车毛重成功" if code == 200 else msg, results)
except Exception as e:
logger.error(traceback.format_exc())
return responseMsg(500, str(e), {}) # 500 内部错误。
@socketio.on('connect')
def ws_connect():
print('Client connected')
send('You are connected!', broadcast=True)
@socketio.on('disconnect')
def ws_disconnect():
print('Client disconnected')
@socketio.on('message')
def handle_message(msg):
print('Received message: ' + msg)
emit('message', "此接口已失效", broadcast=True)
return
# plc = PLC_Tool(config.ip_address, config.slot)
# if msg == "open_report":
# success, msg, results = plc.batch_write_tag(
# [("JZ_B_WRITE[24].19", 1), ("AI.Car_festival2", 77), ("AI.load[1]", 77)])
# if not success:
# emit('message', msg, broadcast=True)
# else:
# if success > 1:
# logger.warn("%s result: %s" % (msg, str(results)))
# emit('message', msg, broadcast=True)
# else:
# emit('message', "ok", broadcast=True)
# elif msg == "close_report":
# success, msg, results = plc.batch_write_tag(
# [("JZ_B_WRITE[24].19", 0), ("AI.Car_festival2", 0), ("AI.load[1]", 0)])
# if not success:
# return responseMsg(412, msg, {})
# else:
# if success > 1:
# logger.warn("%s result: %s" % (msg, str(results)))
# emit('message', msg, broadcast=True)
# else:
# emit('message', "ok", broadcast=True)
# else:
# emit('message', msg, broadcast=True)
@socketio.on('json')
def handle_json(json):
print('received json: ' + str(json))
# def read_plc_thread():
# socketio.detectService.interval_read_PLC_up_web()
#
#
# def web_thread():
# socketio.run(app, host='0.0.0.0', port=config.serverPort, debug=False)
def runWebSocket():
socketio.run(app, host='0.0.0.0', port=config.server_port, debug=False)
def runReadPLC():
while True:
try:
detect.interval_read_PLC_up_web()
except Exception as e:
logger.error("数据 读取&上传 异常：%s", traceback.print_exc())
continue
if __name__ == '__main__':
# --- 运行方式一 ---
# t = Thread(target=read_plc_thread)
# t.daemon = True
# t.start()
#
# server = pywsgi.WSGIServer(('0.0.0.0', serverPort), app)
# server.serve_forever()
# --- 运行方式二 ---
pool = eventlet.GreenPool()
pool.spawn_n(runWebSocket)
pool.spawn_n(runReadPLC)
pool.waitall()

32
config/config.ini Normal file
View File

@ -0,0 +1,32 @@
[server]
server_port = 7010
track_num = 2
[log]
log_path = ./logs
stdout_log_level = INFO
file_log_level = INFO
[plc]
ip_address = 10.227.181.106
slot = 0
timeout = 0.9
interval_plc = 6000
[upload]
upload_type = 1
login_web_url = http://192.168.2.167:20004/api/blade-auth/oauth/token
upload_pinch_coal_url = http://xxxx
upload_gross_weight_url = http://xxxx
upload_load_type_url = http://xxxx
upload_device_type_url = http://192.168.2.167:20004/api/train/train-device-action/api-save
username = guest_01
password = 0a1688bce402a5f6a4279abbfc89c203
authorization = Y2xpZW50X2VudGVycHJpc2U6Y2xpZW50X2VudGVycHJpc2Vfc2VjcmV0
[real_loading]
loading_mark = 857
[other]
carriage_num_max = 149

181
config/plc_tags_config.ini Normal file
View File

@ -0,0 +1,181 @@
[loading_1]
# 报表开启状态
report_open_type= JZ_B_WRITE[24].19
# 正在装车的车节号（配煤完成后+1）
loading_carriage_order= AI.NumberOfCarSections
# 定量仓实时重量
real_weight= JZ_R_READ[33]
# 定量仓闸板打开
gate_open_type= JZ_B_READ[12].9
# 开始配煤
start_coal_blending_type= JZ_B_READ[12].10
# 配煤完成
completion_coal_blending_type= JZ_B_READ[12].11
# 总车节数（写）
total_number_carriage= AI.Car_festival2
# 装车系统启动
load_system_open= JZ_B_READ[1].0
# 预读毛重标志
pre_reading_gross= JZ_B_READ[12].15
# 毛重重读
gross_weight_refresh = PMCD
[loading_2]
# 报表开启状态
report_open_type= JZ_B_WRITE[25].19
# 正在装车的车节号（配煤完成后+1）
loading_carriage_order= L2Numbercar
# 定量仓实时重量
real_weight= JZ_R_READ[34]
# 定量仓闸板打开
gate_open_type= JZ_B_READ[13].9
# 开始配煤
start_coal_blending_type= JZ_B_READ[13].10
# 配煤完成
completion_coal_blending_type= JZ_B_READ[13].11
# 总车节数（写）
total_number_carriage= AI.Car_festival1
# 装车系统启动
load_system_open= JZ_B_READ[13].0
# 预读毛重标志
pre_reading_gross= JZ_B_READ[13].15
# 配煤重读值
gross_weight_refresh= PMCD2
[array_tags_1]
# 掐煤量集合
pinch_coal_weight= JZ_R_READ_QM(1~149)
# 毛重集合
gross_weight= JZ_ConfiglWeight_R_READ(1~149)
# 预装量集合
pre_load_weight= AI.load(1~149)
[array_tags_2]
# 掐煤量集合
pinch_coal_weight= JZ_R_READ_QM2(1~149)
# 毛重集合
gross_weight= JZ_ConfiglWeight_R_READ2(1~149)
# 预装量集合
pre_load_weight= PreInstalledQuantity(1~149)
[belt]
# 7016皮带电机运行
7016= I[38].10
# 7117皮带电机运行指示
7117= I[81].1
# 7025皮带电机运行指示
7025= I[32].1
# 7129皮带电机运行指示
7129= I[82].1
# 7130皮带电机运行指示
7130= I[83].1
# 7045皮带电机运行指示
7045= I[33].1
# 7046皮带电机运行指示
7046= I[34].1
# 7061B皮带电机运行指示
7061B= I[35].1
# 7071皮带电机运行指示
7071= I[90].13
# 7150皮带电机运行指示
7150= I[80].1
[feeder]
# 7003 给煤机运行
7003= I[67].15
# 7004 给煤机运行
7004= I[66].10
# 7005 给煤机运行
7005= I[65].4
# 7006 给煤机运行
7006= I[67].1
# 7007 给煤机运行
7007= I[65].7
# 7008 给煤机运行
7008= I[67].4
# 7009 给煤机运行
7009= I[65].10
# 7010 给煤机VFD运行
7010= I[67].13
# 7011 给煤机运行
7011= I[66].1
# 7012 给煤机运行
7012= I[67].10
# 7013 给煤机运行
7013= I[66].4
# 7014 给煤机运行
7014= I[68].1
# 7015 给煤机运行
7015= I[66].7
# 原煤仓给煤机7031运行指示
7031= I[36].1
# 原煤仓给煤机7032运行指示
7032= I[37].1
# 原煤仓给煤机7033运行指示
7033= I[38].1
# 原煤仓给煤机7034运行指示
7034= I[39].1
# 原煤仓给煤机7035变频器运行指示
7035= I[36].13
# 原煤仓给煤机7036运行指示
7036= I[37].4
# 原煤仓给煤机7037运行指示
7037= I[38].4
# 原煤仓给煤机7038运行指示
7038= I[39].4
# 原煤仓给煤机7039运行指示
7039= I[36].7
# 原煤仓给煤机7040变频器运行指示
7040= I[37].11
# 原煤仓给煤机7041运行指示
7041= I[38].7
# 原煤仓给煤机7042运行指示
7042= I[39].7
# 原煤仓给煤机7043运行指示
7043= I[36].10
# 原煤仓给煤机7044运行指示
7044= I[37].13
# 7106 给煤机运行
7106= I[84].1
# 7107 给煤机运行
7107= I[84].4
# 7108 给煤机运行
7108= I[84].7
# 7109给煤机变频器运行
7109= I[84].11
# 7110 给煤机运行
7110= I[84].14
# 7111 给煤机运行
7111= I[85].1
# 7112给煤机变频器运行
7112= I[85].5
# 7113 给煤机运行
7113= I[85].8
# 7114 给煤机运行
7114= I[85].11
# 7115 给煤机运行
7115= I[85].14
# 7116 给煤机运行
7116= I[88].7
# 7118 给煤机运行
7118= I[86].1
# 7119 给煤机运行
7119= I[86].4
# 7120给煤机变频器运行
7120= I[86].7
# 7121 给煤机运行
7121= I[86].11
# 7122 给煤机运行
7122= I[86].14
# 7123给煤机变频器运行
7123= I[87].1
# 7124 给煤机运行
7124= I[87].5
# 7125 给煤机运行
7125= I[87].8
# 7126 给煤机运行
7126= I[87].11
# 7127 给煤机运行
7127= I[87].14
# 7128 给煤机运行
7128= I[88].2

30
requirements.txt Normal file
View File

@ -0,0 +1,30 @@
altgraph==0.17.3
APScheduler==3.10.4
backports.zoneinfo==0.2.1
bidict==0.22.0
click==8.1.7
dnspython==1.15.0
dnspython3==1.15.0
eventlet==0.33.2
Flask==2.2.5
Flask-SocketIO==5.3.3
gevent==22.10.2
greenlet==2.0.1
importlib-metadata==6.7.0
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.2
pyinstaller==5.13.2
pyinstaller-hooks-contrib==2023.8
pylogix==0.8.14
python-engineio==4.7.0
python-socketio==5.8.0
pytz==2022.6
six==1.16.0
typing_extensions==4.7.1
tzlocal==5.0.1
Werkzeug==2.2.3
zipp==3.15.0
zope.event==4.5.0
zope.interface==6.
requests~=2.31.0

680
service/DetectService.py Normal file
View File

@ -0,0 +1,680 @@
import time
from service.interfaceToWeb import interfaceToWeb
from service.TagsList import TagsList, plc_info
from utils.Tool import logger, config, PLC_Tool
plcInfo = plc_info()
# 定量仓 实时状态 0-等待卸料、1-卸料、2-给料、3-等待操作台启动、4-读取重量、5-停止装载、6-已完成
houseStatus: int = -1
def writePLC(ip: str, slot: int = 0, tags=None):
if tags is None:
return False, "参数tags未写入需要修改的标签点位"
if not isinstance(tags, list):
return False, "参数tags应该是一个[{key:value},{key2:value2}]的格式"
if len(tags) < 1:
return False, "参数tags未写入需要修改的标签点位"
plc = PLC_Tool(ip, slot)
plc.batch_write_tag(tags)
return True
def read_loading_tags(tagsList, plc):
"""
读取【所有股道】PLC装车实时点位
:param tagsList: 点位列表
:param plc: PLC读取对象
:return: 错误信息
"""
if len(tagsList.getLoading_Tags()) < config.track_num:
logger.error("股道数量与PLC点位套数不一致！")
return False
for track in range(config.track_num):
logger.debug("")
logger.debug("------ 股道%d ------", track + 1)
loadingTagsNamelist = [loading_Tag_list[0] for loading_Tag_list in tagsList.getLoading_Tags()[track]]
loadingTagslist = [loading_Tag_list[1] for loading_Tag_list in tagsList.getLoading_Tags()[track]]
success, msg, tags_value = plc.get_tags(loadingTagslist)
if success:
index: int = 0
tag_dict = {}
for tag in tags_value:
# if loadingTagsNamelist[index] == "pre_reading_gross" \
# and len(plcInfo.loadingDict_List) > track:
# print("---", loadingTagsNamelist[index], plcInfo.loadingDict_List[track][loadingTagsNamelist[index]] != tag["value"], (tag["value"] == 1))
# 完成配煤时，记录标志，准备读取毛重(读取两次，一次是配煤完成；一次是配煤完成后开始放料)
if loadingTagsNamelist[index] == "pre_reading_gross" \
and len(plcInfo.loadingDict_List) > track \
and loadingTagsNamelist[index] in plcInfo.loadingDict_List[track].keys() \
and plcInfo.loadingDict_List[track][loadingTagsNamelist[index]] != tag["value"] \
and tag["value"] == 1:
tag_dict["read_gross_weight_%d_type_1" % (track + 1)] = True
if loadingTagsNamelist[index] == "gate_open_type" \
and len(plcInfo.loadingDict_List) > track \
and loadingTagsNamelist[index] in plcInfo.loadingDict_List[track].keys() \
and plcInfo.loadingDict_List[track][loadingTagsNamelist[index]] != tag["value"] \
and tag["value"] == 1:
tag_dict["read_gross_weight_%d_type_2" % (track + 1)] = True
# 开始配煤时，记录标志，准备读取掐煤量
# 注意：此处之前用 completion_coal_blending_type 从true变为false判断该读掐煤量，后发现PLC在点击“配煤重读”时，此值不准确跳动
# 故 改为 start_coal_blending_type 从false变为true时读取；效果会比前者稍慢
if loadingTagsNamelist[index] == "start_coal_blending_type" \
and len(plcInfo.loadingDict_List) > track \
and loadingTagsNamelist[index] in plcInfo.loadingDict_List[track].keys() \
and plcInfo.loadingDict_List[track][loadingTagsNamelist[index]] != tag["value"] \
and tag["value"] == 0:
tag_dict["read_pinch_coal_quantity_%d_type" % (track + 1)] = True
tag_dict[loadingTagsNamelist[index]] = tag["value"]
logger.debug("股道%d- %s : %s", (track + 1), loadingTagsNamelist[index], tag["value"])
index += 1
if len(plcInfo.loadingDict_List) > track:
plcInfo.loadingDict_List[track] = tag_dict
else:
plcInfo.loadingDict_List.append(tag_dict)
else:
logger.error(msg)
return False
pinch_coal_tag = tagsList.getPinchCoalQuantity_Tag()[track]
if len(pinch_coal_tag) == 0:
logger.error("股道%d- 掐煤标签读取失败！请检查配置文件！", track + 1)
return False
success, msg, tags_value = plc.get_array(pinch_coal_tag[1], int(pinch_coal_tag[3]))
if success:
if len(plcInfo.pinchCoalQuantityList_List) > track:
plcInfo.pinchCoalQuantityList_List[track] = tags_value["value"][int(pinch_coal_tag[2]):]
else:
plcInfo.pinchCoalQuantityList_List.append(tags_value["value"][int(pinch_coal_tag[2]):])
logger.debug("股道%d- %s : %s", (track + 1), pinch_coal_tag[1], tags_value["value"][int(pinch_coal_tag[2]):])
else:
logger.error(msg)
return False
gross_weight_tag = tagsList.getGrossWeight_Tags()[track]
if len(gross_weight_tag) == 0:
logger.error("股道%d- 毛重标签读取失败！请检查配置文件！", track + 1)
return False
success, msg, tags_value = plc.get_array(gross_weight_tag[1], int(gross_weight_tag[3]))
if success:
if len(plcInfo.grossWeightList_List) > track:
plcInfo.grossWeightList_List[track] = tags_value["value"][int(gross_weight_tag[2]):]
else:
plcInfo.grossWeightList_List.append(tags_value["value"][int(gross_weight_tag[2]):])
logger.debug("股道%d- %s : %s", (track + 1), pinch_coal_tag[1], tags_value["value"][int(gross_weight_tag[2]):])
else:
logger.error(msg)
return False
return True
def read_preload_tags(tagsList, plc):
"""
读取PLC预装量参数
:param tagsList: 点位列表
:param plc: PLC读取对象
:return: 错误信息
"""
pre_load_weight_Tags = tagsList.getPreLoadWeight_Tags()
for track in range(len(pre_load_weight_Tags)):
if len(pre_load_weight_Tags[track]) == 0:
logger.error("股道%d- 预装量标签读取失败！请检查配置文件！", track + 1)
return False
success, msg, tags_value = plc.get_array(pre_load_weight_Tags[track][1], pre_load_weight_Tags[track][3])
if success:
plcInfo.preLoadWeightList_List.append(tags_value["value"][pre_load_weight_Tags[track][2]:])
else:
logger.error(msg)
return msg
def read_belt_tags(tagsList, plc):
"""
读取PLC 皮带启动状态点位
:param tagsList: 点位列表
:param plc: PLC读取对象
:return: 错误信息
"""
tags_list_name = [tag_list[0] for tag_list in tagsList.getBeltStartTags()]
tags_list = [tag_list[1] for tag_list in tagsList.getBeltStartTags()]
# logger.warn(list(tags_dict.keys()))
success, msg, tags_value = plc.get_tags(tags_list)
if success:
type_change = []
index: int = 0
for tag in tags_value:
if plcInfo.belt_start_dict.get(tags_list_name[index], None) != tag["value"]:
type_change.append({"deviceType": 1,
"deviceCode": tags_list_name[index],
"actionType": int(tag["value"]),
"actionTime": time.strftime('%Y-%m-%d %H:%M:%S')})
plcInfo.belt_start_dict[tags_list_name[index]] = tag["value"]
# logger.debug("%s : %s", tags_list_name[index], tag["value"])
index += 1
else:
logger.error(msg)
return ""
return type_change
def read_feeder_tags(tagsList, plc):
"""
读取PLC 给煤机启动状态点位
:param tagsList: 点位列表
:param plc: PLC读取对象
:return: 错误信息
"""
tags_list_name = [tag_list[0] for tag_list in tagsList.getFeederStartTags()]
tags_list = [tag_list[1] for tag_list in tagsList.getFeederStartTags()]
# logger.warn(list(tags_dict.keys()))
success, msg, tags_value = plc.get_tags(tags_list)
if success:
type_change = []
index: int = 0
for tag in tags_value:
if plcInfo.feeder_start_dict.get(tags_list_name[index], None) != tag["value"]:
type_change.append({"deviceType": 2,
"deviceCode": tags_list_name[index],
"actionType": int(tag["value"]),
"actionTime": time.strftime('%Y-%m-%d %H:%M:%S')})
plcInfo.feeder_start_dict[tags_list_name[index]] = tag["value"]
# logger.debug("%s : %s", tags_list_name[index], tag["value"])
index += 1
else:
logger.error(msg)
return ""
return type_change
def readPLC(plc):
"""
读取PLC的点位
:param plc: 连接的PLC对象
:return:
"""
tagsList = TagsList()
read_result = read_loading_tags(tagsList, plc)
if not read_result:
return False, []
belt_change_type = read_belt_tags(tagsList, plc)
feeder_change_type = read_feeder_tags(tagsList, plc)
return True, [*belt_change_type, *feeder_change_type]
def upload_load_type(loadingDict: dict):
statusList = [
"配煤完成,等待卸料",
"卸料",
"给料",
"等待操作台启动",
"读取重量",
"停止装载",
"已完成"
]
if loadingDict.get("completion_coal_blending_type", False) is True \
and loadingDict.get("gate_open_type", False) is False:
# 配煤完成 并且 闸板未打开 此为 等待卸料
return 0, statusList[0]
elif loadingDict.get("completion_coal_blending_type", False) is True \
and loadingDict.get("gate_open_type", False) is True:
return 1, statusList[1]
elif loadingDict.get("start_coal_blending_type", False) is True:
return 2, statusList[2]
elif loadingDict.get("load_system_open", False) is False \
and loadingDict.get("report_open_type", False) is True:
return 5, statusList[5]
elif loadingDict.get("report_open_type", False) is False:
return 6, statusList[6]
else:
return -1, ""
def interval_read_PLC_up_web():
"""
循环读取PLC数据 根据条件上传至WEB
:return:
"""
plc = PLC_Tool(config.ip_address, config.slot)
interface = interfaceToWeb()
login_result, token_type, access_token = interface.loginWeb(config.login_web_url, config.username,
config.password, config.authorization)
while True:
up_result = ""
start_time = time.time()
load_result, flag_changeds = readPLC(plc)
end_time = time.time()
# print(end_time, "运行时长：%3.f \n" % (end_time - start_time))
if not load_result:
time.sleep(30)
continue
if login_result is not "":
logger.warn(login_result)
return
for track in range(config.track_num):
loadingDict = plcInfo.loadingDict_List[track]
grossWeightList = plcInfo.grossWeightList_List[track]
pinchCoalQuantityList = plcInfo.pinchCoalQuantityList_List[track]
if loadingDict.get("report_open_type", False) is True:
logger.debug("股道%d- 判定上传掐煤-> %s", track+1,
(loadingDict.get("read_pinch_coal_quantity_%d_type" % (track+1), False) is True))
logger.debug("股道%d- 判定第1次上传毛重-> %s", track+1,
(loadingDict.get("read_gross_weight_%d_type_1" % (track + 1), False) is True))
logger.debug("股道%d- 判定第2次上传毛重-> %s", track+1,
(loadingDict.get("read_gross_weight_%d_type_2" % (track + 1), False) is True))
if loadingDict.get("read_pinch_coal_quantity_%d_type" % (track+1), False) is True \
or (
loadingDict.get("loading_carriage_order", -1) == loadingDict.get("total_number_carriage", -2)
and loadingDict.get("load_system_open", True) is False
and loadingDict.get("completion_coal_blending_type", True) is False):
loadingDict["read_pinch_coal_quantity_%d_type" % (track+1)] = False
if loadingDict["loading_carriage_order"] == 0:
logger.warn("获取到掐煤量，但是车节号为0，取消上传")
continue
if config.upload_type == 1:
up_result = interface.upload_pinch_coal(config.upload_pinch_coal_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
loadingDict["loading_carriage_order"],
pinchCoalQuantityList[
loadingDict["loading_carriage_order"]-1],
str(track+1))
logger.info(
"股道%d- 获取到第%d节车厢的掐煤数：%2.f 千克" % (
(track+1), loadingDict["loading_carriage_order"],
pinchCoalQuantityList[loadingDict["loading_carriage_order"]-1]))
if loadingDict.get("read_gross_weight_%d_type_1" % (track + 1), False) is True:
loadingDict["read_gross_weight_%d_type_1" % (track + 1)] = False
if loadingDict["loading_carriage_order"] == 0:
logger.warn("获取到装车毛重，但是车节号为0，取消上传")
continue
if config.upload_type == 1:
up_result = interface.upload_gross_weight(config.upload_gross_weight_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
loadingDict["loading_carriage_order"],
float(loadingDict.get("real_weight",
0)) * 1000.0,
str(track+1))
logger.info(
"股道%d- 获取到第%d节车厢的预计毛重值：%2.f 千克" % (
(track+1), loadingDict["loading_carriage_order"],
float(loadingDict.get("real_weight", 0)) * 1000.0))
if loadingDict.get("read_gross_weight_%d_type_2" % (track + 1), False) is True \
and loadingDict.get("completion_coal_blending_type", False) is True:
loadingDict["read_gross_weight_%d_type_2" % (track + 1)] = False
if loadingDict["loading_carriage_order"] == 0:
logger.warn("获取到装车毛重，但是车节号为0，取消上传")
continue
if config.upload_type == 1:
up_result = interface.upload_gross_weight(config.upload_gross_weight_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
loadingDict["loading_carriage_order"],
grossWeightList[
loadingDict["loading_carriage_order"]-1],
str(track+1))
logger.info(
"股道%d- 获取到第%d节车厢的实际毛重值：%2.f 千克" % (
(track+1), loadingDict["loading_carriage_order"],
grossWeightList[loadingDict["loading_carriage_order"]-1]))
if loadingDict.get("gross_weight_refresh", 0) > 0:
if loadingDict["loading_carriage_order"] == 0:
logger.warn("获取到配煤重读毛重，但是车节号为0，取消上传")
continue
if config.upload_type == 1:
up_result = interface.upload_gross_weight(config.upload_gross_weight_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
loadingDict["loading_carriage_order"],
loadingDict.get("gross_weight_refresh", 0),
str(track+1))
logger.info(
"股道%d- 获取到第%d节车厢的配煤重读毛重值：%2.f 千克" % (
(track+1), loadingDict["loading_carriage_order"],
loadingDict.get("gross_weight_refresh", 0)))
plc = PLC_Tool(config.ip_address, config.slot)
grossrefresh_Tag = ""
for grossrefresh_l in TagsList().getLoading_Tags()[track]:
if grossrefresh_l[0] == "gross_weight_refresh" and len(grossrefresh_l) > 1:
grossrefresh_Tag = grossrefresh_l[1]
if grossrefresh_Tag == "":
logger.error("股道%d- 配煤重读点位读取失败 ！！！", (track + 1))
success, msg = plc.write_tag(grossrefresh_Tag, 0)
if not success:
logger.error("股道%d- 配煤重读归零失败 ！！！", (track+1))
if up_result != "":
logger.warn(up_result)
status, status_str = upload_load_type(loadingDict)
if status != -1 and "loading_carriage_order" in loadingDict.keys():
global houseStatus
if houseStatus != status:
houseStatus = status
if config.upload_type == 1:
up_result = interface.update_load_type(config.upload_gross_weight_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
loadingDict["loading_carriage_order"],
houseStatus,
str(track+1))
logger.info(
"股道%d- 装车状态变化：%d-%s" % ((track+1), houseStatus, status_str))
if up_result != "":
logger.warn(up_result)
if len(flag_changeds) > 0:
if config.upload_type == 1:
up_result = interface.upload_device_type(config.upload_device_type_url,
flag_changeds)
logger.info(
"股道%d- 皮带给煤机状态变化：%s" % ((track+1), str(flag_changeds)))
if up_result != "":
logger.warn(up_result)
logger.debug("")
time.sleep(config.interval_plc / 1000)
def report_open(pound_no: str, total_number_carriage: int, pre_load: list):
"""
启动PLC报表
:param pound_no: 股道名
:param total_number_carriage: 总车节数
:param pre_load: 预装量
:return: [响应码:int, 提示信息:str, 结构数据:obj]
"""
over_write = False
old_total_number_carriage = 0
old_pre_load = [0] * config.carriage_num_max
if pound_no is None:
return 412, 'no pound_no', {}
if isinstance(pound_no, str) and not pound_no.isdigit():
return 412, "pound_no should be a int and 0 <= track < 2", {}
if not (0 < int(pound_no) <= 2):
return 412, "pound_no should be a int and 0 < track <= 2", {}
if total_number_carriage == 0:
return 412, 'no total_number_carriage', {}
if isinstance(total_number_carriage, str) and not total_number_carriage.isdigit():
return 412, "total_number_carriage should be a int and > 0", {}
if not pre_load:
return 412, 'no pre_load', {}
if not isinstance(pre_load, list):
return 412, '\"pre_load\" It should be of list type', {}
if not all(isinstance(item, int) and item >= 0 for item in pre_load):
return 412, '\"pre_load\" It should be of list[int,int,...] type and every pre_load >= 0', {}
if int(total_number_carriage) != len(pre_load):
return 412, '\'total_number_carriage\' should be the same length as the list \'pre_load\'', {}
pound_no = int(pound_no)
if pound_no > config.track_num:
return 412, '开启报表的股道号超过当前设置的股道数量，请检查配置', {}
total_number_carriage = int(total_number_carriage)
tagsList = TagsList()
loadingTagsList = tagsList.getLoading_Tags()[pound_no-1]
report_open_type_tag = 0
total_number_carriage_tag = 0
for tag_list in loadingTagsList:
if tag_list[0] == "report_open_type":
report_open_type_tag = tag_list[1]
elif tag_list[0] == "total_number_carriage":
total_number_carriage_tag = tag_list[1]
plc = PLC_Tool(config.ip_address, config.slot)
if "report_open_type" in plcInfo.loadingDict_List[pound_no-1].keys() and plcInfo.loadingDict_List[pound_no-1]["report_open_type"] == 1:
over_write = True
old_total_number_carriage = plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"]
success, msg, tags_value = plc.get_array(tagsList.getPreLoadWeight_Tags()[pound_no-1][1], tagsList.getPreLoadWeight_Tags()[pound_no-1][3])
if success:
old_pre_load = tags_value["value"]
success, msg, results = \
plc.batch_write_tag([(report_open_type_tag, 1),
(total_number_carriage_tag, total_number_carriage)])
if not success:
return 412, msg, {}
else:
if success > 1:
return 417, msg, results
# else:
# return responseMsg(200, msg, results)
# 数字下标0 都不用（同毛重、掐煤量）
pre_load.insert(0, 0.0)
logger.info("股道%d- 总车节数：%d 上传预装量：%s", pound_no, total_number_carriage, str(pre_load))
success, msg = plc.write_array("AI.load", pre_load)
if not success:
return 412, msg, {}
if over_write:
if old_total_number_carriage != total_number_carriage and old_total_number_carriage != 0:
logger.warn(
"股道%d- 【开启PLC报表】疑似更新装车总节数! 输入的车厢总数为%d节，而原PLC中的车厢总数为%d节；当前在装车厢为第%d节",
pound_no,
total_number_carriage, old_total_number_carriage,
plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"])
if old_pre_load != pre_load and old_pre_load != [0] * config.carriage_num_max:
logger.warn(
"【开启PLC报表】疑似更新装车预装量! 输入的车厢总数为%d节" % total_number_carriage)
if plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"] == plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"]:
logger.warn(
"股道%d- 【开启PLC报表】疑似装最后一节时更新装车数据! 输入的车厢总数为%d节，而原PLC中的车厢总数为%d节；当前在装车厢为第%d节",
pound_no,
total_number_carriage, plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"],
plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"])
else:
logger.info("【开启PLC报表】")
return 200, msg, results
def report_close(pound_no: str, total_number_carriage: int):
"""
关闭PLC报表
:param pound_no: 股道名
:param total_number_carriage: 总车节数
:return: [响应码:int, 提示信息:str, 结构数据:obj]
"""
if pound_no is None:
return 412, 'no pound_no', {}
if isinstance(pound_no, str) and not pound_no.isdigit():
return 412, "pound_no should be a int and 0 < track <= 2", {}
if not (0 < int(pound_no) <= 2):
return 412, "pound_no should be a int and 0 < track <= 2", {}
if total_number_carriage == 0:
return 412, 'no total_number_carriage', {}
if (isinstance(total_number_carriage, str) and not total_number_carriage.isdigit()) \
or int(total_number_carriage) <= 0:
return 412, "total_number_carriage should be a int and > 0", {}
pound_no = int(pound_no)
total_number_carriage = int(total_number_carriage)
tagsList = TagsList()
open_report_tags_list = tagsList.getOpenReportTags()
total_number_carriage_tags_list = tagsList.getTotalNumberCarriageTags()
if len(open_report_tags_list) <= (pound_no - 1):
return 412, "股道“开启报表”的点位配置未读到，请检查配置文件", {}
if len(total_number_carriage_tags_list) <= (pound_no - 1):
return 412, "股道“总车节数”的点位配置未读到，请检查配置文件", {}
pre_load_tags_list = tagsList.getPreLoadWeightTags()
if len(pre_load_tags_list) < (pound_no - 1):
return 412, "股道“预装量”的点位配置未读到，请检查配置文件", {}
plc = PLC_Tool(config.ip_address, config.slot)
success, msg, results = plc.batch_write_tag([(open_report_tags_list[pound_no - 1], 0),
(total_number_carriage_tags_list[pound_no - 1], 0)])
if not success:
return 412, msg, {}
else:
if success > 1:
return 417, msg, results
# else:
# return responseMsg(200, msg, results)
pre_load = [0] * config.carriage_num_max
success, msg = plc.write_array(pre_load_tags_list[pound_no - 1], pre_load)
if not success:
return 412, msg, {}
if len(plcInfo.loadingDict_List) < 1:
return 412, "PLC连接失败，数据异常", {}
if total_number_carriage != plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"]:
logger.warn(
"【关闭PLC报表】疑似关闭的非同一列车! 输入的车厢总数为%d节，而实际PLC中的车厢总数为%d节；当前在装车厢为第%d节" % (
total_number_carriage, plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"],
plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"]))
elif plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"] != plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"]:
logger.warn(
"【关闭PLC报表】疑似提前误关报表! 输入的车厢总数为%d节，而实际PLC中的车厢总数为%d节；当前在装车厢为第%d节" % (
total_number_carriage, plcInfo.loadingDict_List[pound_no-1]["total_number_carriage"],
plcInfo.loadingDict_List[pound_no-1]["loading_carriage_order"]))
else:
logger.info("【关闭PLC报表】")
return 200, msg, {"carriageWeightHouse": plcInfo.pinchCoalQuantityList_List[pound_no-1],
"grossWeightHouse": plcInfo.grossWeightList_List[pound_no-1]}
def get_web_token():
"""
请求获取web token
:return:
"""
login_result, token_type, access_token = interfaceToWeb().loginWeb(config.login_web_url, config.username,
config.password, config.authorization)
if login_result is not "":
logger.warn(login_result)
return 412, "请求登录web失败,原因：%s" % login_result, {}
return 200, "", {"token_type": token_type, "access_token": access_token, "token": (token_type + " " + access_token)}
def upload_pinch_coal(carriage_order: int, pinch_coal: float, pound_no: str):
"""
上传掐煤量
:param carriage_order: 车厢序号
:param pinch_coal: 掐煤量
:param pound_no: 股道名
:return: [响应码:int, 提示信息:str, 结构数据:obj]
"""
if carriage_order == 0:
return 412, "no carriage_order", {}
if isinstance(carriage_order, str) and not carriage_order.isdigit() or int(carriage_order) <= 0:
return 412, "\"carriage_order\" It should be of INT type and greater than 0.", {}
if pinch_coal == -1:
return 412, "no pinch_coal", {}
if isinstance(pinch_coal, str) and not pinch_coal.isdigit() or float(pinch_coal) < 0:
return 412, "\"pinch_coal\" It should be of float type and greater than 0.", {}
if pound_no == "":
return 412, "no pound_no", {}
if isinstance(pound_no, str) and not pound_no.isdigit():
return 412, "pound_no should be a int, and 0 < pound_no <= 2 ", {}
if not 0 < int(pound_no) <= config.track_num:
return 412, "pound_no should be a int, and 0 < pound_no <= 2 ", {}
interface = interfaceToWeb()
login_result, token_type, access_token = interface.loginWeb(config.login_web_url, config.username,
config.password, config.authorization)
if login_result is not "":
logger.warn(login_result)
return 412, "请求登录web失败", {}
up_result = interface.upload_pinch_coal(config.upload_pinch_coal_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
int(carriage_order),
float(pinch_coal),
pound_no)
if up_result != "":
return 417, up_result, {}
else:
return 200, "上传成功", {}
def upload_gross_weight(carriage_order: int, gross_weight: float, pound_no: str):
"""
上传毛重
:param carriage_order: 车厢序号
:param gross_weight: 毛重
:param pound_no: 股道名
:return: [响应码:int, 提示信息:str, 结构数据:obj]
"""
if carriage_order == 0:
return 412, "no carriage_order", {}
if isinstance(carriage_order, str) and not carriage_order.isdigit() or int(carriage_order) <= 0:
return 412, "\"carriage_order\" It should be of INT type and greater than 0.", {}
if gross_weight == -1:
return 412, "no pinch_coal", {}
if isinstance(gross_weight, str) and not gross_weight.isdigit() or float(gross_weight) <= 0:
return 412, "\"gross_weight\" It should be of float type and greater than 0.", {}
if pound_no == "":
return 412, "no pound_no", {}
if isinstance(pound_no, str) and not pound_no.isdigit():
return 412, "pound_no should be a int, and 0 < pound_no <= 2 ", {}
if not 0 < int(pound_no) <= config.track_num:
return 412, "pound_no should be a int, and 0 < pound_no <= 2 ", {}
interface = interfaceToWeb()
login_result, token_type, access_token = interface.loginWeb(config.login_web_url, config.username,
config.password, config.authorization)
if login_result is not "":
logger.warn(login_result)
return 412, "请求登录web失败", {}
up_result = interface.upload_gross_weight(config.upload_pinch_coal_url,
time.strftime('%Y-%m-%d %H:%M:%S'),
int(carriage_order),
float(gross_weight),
pound_no)
if up_result != "":
return 417, up_result, {}
else:
return 200, "上传成功", {}

249
service/TagsList.py Normal file
View File

@ -0,0 +1,249 @@
from utils.PlcTagsConfigT import PlcTagsConfigT
class plc_info(object):
def __init__(self):
self.loadingDict_List = []
self.pinchCoalQuantityList_List = []
self.grossWeightList_List = []
self.preLoadWeightList_List = []
self.belt_start_dict = {}
self.feeder_start_dict = {}
class TagsList(object):
"""
"""
def __init__(self):
# 装车点位标签
self.__loading_tags_list = []
self.__loading_tags_dict_list = []
# 集合[掐煤量、毛重、预装量]标签列表
self.__array_tags_list = []
# 皮带标签列表
self.__belt_start_tags_list = []
# 给煤机标签列表
self.__feeder_start_tags_list = []
# 获取多个股道 开启报表点位列表
self.__open_report_tags_list = []
# 获取多个股道 总车节数的点位列表
self.__total_number_carriage_tags_list = []
# 获取多个股道 预装量点位列表
self.__pre_load_weight_tags_list = []
plc_tags_config = PlcTagsConfigT()
self.__loading_tags_list.append(plc_tags_config.loading_1_tags_list)
self.__loading_tags_list.append(plc_tags_config.loading_2_tags_list)
self.__loading_tags_dict_list.append(plc_tags_config.loading_1_tags_dict)
self.__loading_tags_dict_list.append(plc_tags_config.loading_2_tags_dict)
self.__array_tags_list.append(plc_tags_config.array_1_tags_list)
self.__array_tags_list.append(plc_tags_config.array_2_tags_list)
self.__feeder_start_tags_list = plc_tags_config.feeder_tags_list
self.__belt_start_tags_list = plc_tags_config.belt_tags_list
def getLoading_Tags_Dict_List(self):
"""
获取所有股道装车实时相关标签字典
:return: 标签列表 :[ {标签名1:标签1,标签名2:标签2}, {2股道...}]
"""
return self.__loading_tags_dict_list
def getLoading_Tags(self):
"""
获取所有股道装车实时相关标签列表
:return: 标签列表 :[ [[标签名1,标签1],[标签名2,标签2]], [2股道...]]
"""
return self.__loading_tags_list
def getLoading_1_Tags(self):
"""
获取装车实时相关标签列表
:return: 标签列表 :list[[标签名1,标签1],[标签名2,标签2]]
"""
return self.__loading_tags_list[0]
def getLoading_2_Tags(self):
"""
获取装车实时相关标签列表
:return: 标签列表 :list[[标签名1,标签1],[标签名2,标签2]]
"""
return self.__loading_tags_list[1]
def getPinchCoalQuantity_Tag(self):
"""
获取所有股道每节车厢掐煤量标签列表
:return: 标签列表 :[[标签名1, 标签1, 最小下标, 最大下标], [2股道...]]
"""
tag = []
for track_tag_list in self.__array_tags_list:
for tag_list in track_tag_list:
if tag_list[0] == "pinch_coal_weight":
tag.append(tag_list)
return tag
def getPinchCoalQuantity_1_Tag(self):
"""
获取每节车厢掐煤量标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[0]:
if tag_list[0] == "pinch_coal_weight":
return tag_list
return []
def getPinchCoalQuantity_2_Tag(self):
"""
获取每节车厢掐煤量标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[1]:
if tag_list[0] == "pinch_coal_weight":
return tag_list
return []
def getGrossWeight_Tags(self):
"""
获取所有股道每节车厢装车毛重标签列表
:return: 标签列表 :[[标签名1, 标签1, 最小下标, 最大下标], [2股道...]]
"""
tag = []
for track_tag_list in self.__array_tags_list:
for tag_list in track_tag_list:
if tag_list[0] == "gross_weight":
tag.append(tag_list)
return tag
def getGrossWeight_1_Tags(self):
"""
获取每节车厢装车毛重标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[0]:
if tag_list[0] == "gross_weight":
return tag_list
return []
def getGrossWeight_2_Tags(self):
"""
获取每节车厢装车毛重标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[1]:
if tag_list[0] == "gross_weight":
return tag_list
return []
def getPreLoadWeight_Tags(self):
"""
获取所有股道每节车厢预装量标签列表
:return: 标签列表 :[[标签名1, 标签1, 最小下标, 最大下标], [2股道...]]
"""
tag = []
for track_tag_list in self.__array_tags_list:
for tag_list in track_tag_list:
if tag_list[0] == "pre_load_weight":
tag.append(tag_list)
return tag
def getPreLoadWeight_1_Tags(self):
"""
获取每节车厢预装量标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[0]:
if tag_list[0] == "pre_load_weight":
return tag_list
return []
def getPreLoadWeight_2_Tags(self):
"""
获取每节车厢预装量标签列表
:return: 标签列表 :list[标签名1, 标签1, 最小下标, 最大下标]
"""
for tag_list in self.__array_tags_list[1]:
if tag_list[0] == "pre_load_weight":
return tag_list
return []
def getBeltStartTags(self):
"""
获取皮带开始运行状态
:return: [[标签名1,标签1],[标签名2,标签2]]
"""
return self.__belt_start_tags_list
def getFeederStartTags(self):
"""
获取给煤机开始运行状态
:return: [[标签名1,标签1],[标签名2,标签2]]
"""
return self.__feeder_start_tags_list
def getOpenReportTags(self):
"""
获取所有股道的 报表开启状态 标签
:return:
"""
try:
if len(self.__open_report_tags_list) == 0:
for loadingT_List in self.getLoading_Tags():
for tag_name_index in range(len([loadingT[0] for loadingT in loadingT_List])):
if loadingT_List[tag_name_index][0] == "report_open_type":
self.__open_report_tags_list.append(loadingT_List[tag_name_index][1])
except Exception as e:
pass
return self.__open_report_tags_list
def getTotalNumberCarriageTags(self):
"""
获取所有股道的 总车节数 标签
:return:
"""
try:
if len(self.__total_number_carriage_tags_list) == 0:
for loadingT_List in self.getLoading_Tags():
for tag_name_index in range(len([loadingT[0] for loadingT in loadingT_List])):
if loadingT_List[tag_name_index][0] == "total_number_carriage":
self.__total_number_carriage_tags_list.append(loadingT_List[tag_name_index][1])
except Exception as e:
pass
return self.__total_number_carriage_tags_list
def getPreLoadWeightTags(self):
"""
获取所有股道的 预装量 标签
:return:
"""
try:
for array in self.__array_tags_list:
for tag_name_index in range(len(array)):
if array[tag_name_index][0] == "pre_load_weight":
self.__pre_load_weight_tags_list.append(array[tag_name_index][1])
except Exception as e:
pass
return self.__pre_load_weight_tags_list

191
service/interfaceToWeb.py Normal file
View File

@ -0,0 +1,191 @@
import json
import requests
class interfaceToWeb:
def __init__(self):
self.url: str = ""
self.username: str = ""
self.password: str = ""
self.authorization: str = ""
self.token: str = ""
def loginWeb(self, url: str, username: str, password: str, authorization: str):
"""
登录web
:param url: 请求地址
:param username: 用户名
:param password: 密码
:param authorization: 授权
:return:
"""
self.url: str = url
self.username: str = username
self.password: str = password
self.authorization: str = authorization
headers = {'Authorization': "Basic %s" % str(authorization)}
data = {"username": str(username),
"password": str(password),
"tenantId": "000000",
"grant_type": "password"}
try:
res = requests.post(url=url, headers=headers, data=data)
if res.status_code == requests.codes.ok:
jsonData = json.loads(res.text)
if jsonData.get('access_token', None) is not None and jsonData.get('token_type', None) is not None:
self.token = "%s %s" % (jsonData.get('token_type'), jsonData.get('access_token'))
return "", jsonData.get('token_type'), jsonData.get('access_token')
else:
return "【登录web】请求成功，但返回信息中不含。请求信息%s token :%s" % (str(data), str(res.text)), None, None
else:
return "【登录web】请求登录失败，状态码：%d，请求信息：%s， 返回信息：%s" % (
res.status_code, str(data), str(res.text)), None, None
except Exception as e:
return "【登录web】访问登录失败！疑似网络问题！", None, None
def upload_pinch_coal(self, url: str, load_time: str, carriage_order: int,
pinch_coal: float,
track_name: str):
"""
上传掐煤量
:param url: 请求地址
:param load_time: 装车时间
:param carriage_order: 车厢节号
:param pinch_coal: 掐煤量
:param track_name: 股道名
:return: 请求结果:str
"""
# url = "http://localhost:20004/train-manual/train-manual-detail/api-save"
headers = {'Content-Type': 'application/json', 'blade-auth': self.token}
data = {"loadTime": str(load_time),
"carriageOrder": carriage_order,
"carriageWeightHouse": pinch_coal,
# "houseStatus": 4,
"poundNo": track_name}
try:
res = requests.post(url=url, headers=headers, json=data)
if res.status_code == requests.codes.ok:
jsonData = json.loads(res.text)
if not jsonData['success']:
if res.status_code == 401:
self.loginWeb(self.url, self.username, self.password, self.authorization)
if self.upload_pinch_coal(url, load_time, carriage_order, pinch_coal, track_name) is "":
return ""
return "【上传掐煤量】联通请求，但返回失败：%s，请求信息：%s" % (jsonData['msg'], str(data))
else:
return ""
else:
return "【上传掐煤量】请求失败，状态码：%d，请求信息：%s， 返回信息：%s" % (
res.status_code, str(data), str(res.text))
except Exception as e:
return "【上传掐煤量】访问接口失败！疑似网络问题！"
def upload_gross_weight(self, url: str, load_time: str, carriage_order: int,
gross_weight: float,
track_name: str):
"""
上传装车毛重
:param url: 请求地址
:param load_time: 装车时间
:param carriage_order: 车厢节号
:param gross_weight: 掐煤量
:param track_name: 股道名
:return: 请求结果:str
"""
# url = "http://localhost:20004/train-manual/train-manual-detail/api-save"
headers = {'Content-Type': 'application/json', 'blade-auth': self.token}
data = {"loadTime": str(load_time),
"carriageOrder": carriage_order,
"grossWeightHouse": gross_weight,
"houseStatus": 4,
"poundNo": track_name}
try:
res = requests.post(url=url, headers=headers, json=data)
if res.status_code == requests.codes.ok:
jsonData = json.loads(res.text)
if not jsonData['success']:
if res.status_code == 401:
self.loginWeb(self.url, self.username, self.password, self.authorization)
if self.upload_gross_weight(url, load_time, carriage_order, gross_weight, track_name) is "":
return ""
return "【上传装车毛重】联通请求，但返回失败：%s，请求信息：%s" % (jsonData['msg'], str(data))
else:
return ""
else:
return "【上传装车毛重】请求失败，状态码：%d，请求信息：%s， 返回信息：%s" % (
res.status_code, str(data), str(res.text))
except Exception as e:
return "【上传装车毛重】访问接口失败！疑似网络问题！"
def update_load_type(self, url: str, load_time: str, carriage_order: int,
house_status: int,
track_name: str):
"""
修改装车状态
:param url: 请求地址
:param load_time: 装车时间
:param carriage_order: 车厢节号
:param house_status: 装车状态
:param track_name: 股道名
:return: 请求结果:str
"""
headers = {'Content-Type': 'application/json', 'blade-auth': self.token}
data = {"loadTime": str(load_time),
"carriageOrder": carriage_order,
"houseStatus": house_status,
"poundNo": track_name}
try:
res = requests.post(url=url, headers=headers, json=data)
if res.status_code == requests.codes.ok:
jsonData = json.loads(res.text)
if not jsonData['success']:
if res.status_code == 401:
self.loginWeb(self.url, self.username, self.password, self.authorization)
if self.update_load_type(url, load_time, carriage_order, house_status, track_name) is "":
return ""
return "【修改装车状态】联通请求，但返回失败：%s，请求信息：%s" % (jsonData['msg'], str(data))
else:
return ""
else:
return "【修改装车状态】请求失败，状态码：%d，请求信息：%s， 返回信息：%s" % (
res.status_code, str(data), str(res.text))
except Exception as e:
return "【修改装车状态】访问接口失败！疑似网络问题！"
def upload_device_type(self, url: str, data):
"""
:param url: 请求地址
:param data: 设备状态集合
:return: 请求结果:str
"""
headers = {'Content-Type': 'application/json', 'blade-auth': self.token}
try:
res = requests.post(url=url, headers=headers, json=data)
if res.status_code == requests.codes.ok:
jsonData = json.loads(res.text)
if not jsonData['success']:
if res.status_code == 401:
self.loginWeb(self.url, self.username, self.password, self.authorization)
if self.upload_device_type(url, data) is "":
return ""
return "【上传设备状态】联通请求，但返回失败：%s，请求信息：%s" % (jsonData['msg'], str(data))
else:
return ""
else:
return "【上传设备状态】请求失败，状态码：%d，请求信息：%s， 返回信息：%s" % (
res.status_code, str(data), str(res.text))
except Exception as e:
return "【上传设备状态】访问接口失败！疑似网络问题！"

118
utils/ConfigT.py Normal file
View File

@ -0,0 +1,118 @@
import configparser
import json
config_ini = configparser.ConfigParser()
class ConfigT:
def __init__(self, config_path):
config_ini.read(config_path, encoding="utf-8")
self.config_path = config_path
self.server_port = None
self.track_num = None
self.log_path = None
self.stdout_log_level = None
self.file_log_level = None
self.ip_address = None
self.slot = None
self.timeout = None
self.interval_plc = None
self.upload_type = None
self.upload_pinch_coal_url = None
self.upload_gross_weight_url = None
self.upload_device_type_url = None
self.login_web_url = None
self.username = None
self.password = None
self.authorization = None
self.loading_mark = None
self.carriage_num_max = None
self.initConfig()
def initConfig(self):
self.server_port = config_ini.getint('server', 'server_port')
# 股道名称
self.track_num = config_ini.getint('server', 'track_num')
# 日志地址
self.log_path = config_ini.get('log', 'log_path')
# 打印日志的存储级别
self.stdout_log_level = config_ini.get('log', 'stdout_log_level')
# 文件日志的存储级别
self.file_log_level = config_ini.get('log', 'file_log_level')
# PLC 的IP地址
self.ip_address = config_ini.get('plc', 'ip_address')
# PLC 插槽
self.slot = config_ini.getint('plc', 'slot')
# 访问超时时间 秒为单位
self.timeout = config_ini.getfloat('plc', 'timeout')
# 读取PLC的循环间隔
self.interval_plc = config_ini.getint('plc', 'interval_plc')
# 上传状态
self.upload_type = config_ini.getint('upload', 'upload_type')
# 登录web
self.login_web_url = config_ini.get('upload', 'login_web_url')
# 掐煤量上传地址
self.upload_pinch_coal_url = config_ini.get('upload', 'upload_pinch_coal_url')
# 毛重上传地址
self.upload_gross_weight_url = config_ini.get('upload', 'upload_gross_weight_url')
# 皮带、给煤机启停状态上传地址
self.upload_device_type_url = config_ini.get('upload', 'upload_device_type_url')
# 登录用户名
self.username = config_ini.get('upload', 'username')
# 登录密码
self.password = config_ini.get('upload', 'password')
# 登录授权码
self.authorization = config_ini.get('upload', 'authorization')
# 装车车标记号
self.loading_mark = config_ini.get('real_loading', 'loading_mark')
# 最大车厢数量
self.carriage_num_max = config_ini.getint('other', 'carriage_num_max')
def getConfig(self):
return json.dumps(self.__dict__)
def updateConfig(self, option, value):
try:
if option in {"server_port", "track_num"}:
if option == "track_num" and not 0 < int(value) <= 2:
return "配置项 %s 应该0<%s<=2，而本次设置内容为 %s" % (option, option, value)
config_ini.set("server", option, value)
elif option in {"log_path", "stdout_log_level", "file_log_level"}:
config_ini.set("log", option, value)
elif option in {"ip_address", "slot", "timeout", "interval_plc"}:
config_ini.set("plc", option, value)
elif option in {"upload_type",
"login_web_url",
"upload_pinch_coal_url",
"upload_gross_weight_url",
"upload_device_type_url",
"username",
"password",
"authorization"}:
config_ini.set("upload", option, value)
elif option in {"loading_mark"}:
config_ini.set("real_loading", option, value)
elif option in {"carriage_num_max"}:
config_ini.set("other", option, value)
else:
return "未找到此配置项 {%s: %s}" % (option, value)
config_ini.write(open(self.config_path, 'w'))
self.initConfig()
return ""
except Exception as e:
print("更改配置参数异常！%s" % str(e))
return "更改配置参数异常！%s" % str(e)

278
utils/ConnectPLC.py Normal file
View File

@ -0,0 +1,278 @@
import traceback
import re
from pylogix import PLC
class PLC_Tool:
def __init__(self, plcIp: str, plcSlot: int = 0, timeOut: float = 5):
self.plcIp = plcIp
self.plcSlot = plcSlot
self.timeOut = timeOut
self.comm = None
self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut)
def __del__(self):
self.comm.Close()
def __isIP(self, ip: str):
"""
判断是否是IP地址
:param ip:
:return: bool
"""
pattern = r'^((?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]\d|\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]\d|\d)$'
if re.match(pattern, ip):
return True
else:
return False
def connect(self):
self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut)
def close(self):
self.comm.Close()
def batch_write_tag(self, tags: list):
"""
Write tags to PLC . 批量写入标签
:param tags: [("AI.speed", 10.32),("AI.start", 1)]
:return: [int:(0:失败，1:成功, 2:成功部分), str:异常信息, list: 修改的标签tuple]
"""
connect_count: int = 0
try:
if self.plcIp is None:
return 0, "请输入PLC的IP地址", None
if not self.__isIP(self.plcIp):
return 0, "PLC的IP地址格式异常", None
if tags is None:
return 0, "参数tags未写入需要修改的标签点位", None
if not isinstance(tags, list):
return 0, "参数tags应该是一个[{key:value},{key2:value2}]的格式", None
if len(tags) == 0:
return 0, "请输入标签tags:list", None
for tag in tags:
if not isinstance(tag, tuple):
return 0, "输入的参数格式不对，请参照：[(\"AI.speed\", 10.32),(\"AI.start\", 1)]", None
print("需要写入的数据：", tags)
rets = self.comm.Write(tags)
hasNone = False
results = []
for ret in rets:
results.append(
{"tagName": ret.TagName,
"value": ret.Value,
"status": ret.Status
})
if ret.Status != "Success":
hasNone = True
if hasNone:
return 2, "存在写入失败的标签", results
else:
return 1, "", results
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return 0, str(io_error), None
connect_count += 1
self.connect()
self.batch_write_tag(tags)
except Exception as e:
print(traceback.format_exc())
return 0, str(e), None
def write_tag(self, tag: str, value: [str, int, float]):
"""
Write tags to PLC . 批量写入标签
:param tag: [("AI.speed", 10.32),("AI.start", 1)]
:param value: type is [str|int|float]
:return: [bool, str:异常信息, list: 修改的标签tuple]
"""
connect_count: int = 0
try:
if self.plcIp is None:
return False, "请输入PLC的IP地址"
if not self.__isIP(self.plcIp):
return False, "PLC的IP地址格式异常"
if len(tag) == 0:
return False, "请输入标签tag:str"
# if len(value) == 0:
# return False, "请输入标签'%s'的新值value:str" % tag
print("需要写入的数据 %s:%s" % (tag, str(value)))
ret = self.comm.Write(tag, value)
if ret.Status != "Success":
return False, "存在写入失败的标签"
else:
return True, ""
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return False, str(io_error)
connect_count += 1
self.connect()
self.write_tag(tag, value)
except Exception as e:
print(traceback.format_exc())
return False, str(e)
def write_array(self, tag: str, array: list):
"""
写入数组
:param tag: 标签的名字，例如AI.speed[12]
:param array:具体数值
:return: [bool, str:异常信息]
"""
connect_count: int = 0
try:
if self.plcIp is None:
return False, "请输入PLC的IP地址"
if not self.__isIP(self.plcIp):
return False, "PLC的IP地址格式异常"
if len(tag) == 0:
return False, "请输入标签tag:str"
if len(array) == 0:
return False, "请输入要写入的参数值"
ret = self.comm.Write(tag, array)
# results = [{"tagName": ret.TagName,
# "value": ret.Value,
# "status": ret.Status
# }]
if ret.Status != "Success":
return False, "数组信息写入失败"
return True, ""
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return False, str(io_error)
connect_count += 1
self.connect()
self.write_array(tag, array)
except Exception as e:
print(traceback.format_exc())
return False, str(e)
def get_tags(self, tags: [str, list]):
"""
获取标签或者标签列表
:param tags: 目标标签列表 [str|list]
:return: [bool, str, list]
"""
results = []
connect_count: int = 0
try:
if self.plcIp is None:
return False, "请输入PLC的IP地址", None
if not self.__isIP(self.plcIp):
return False, "PLC的IP地址格式异常", None
if len(tags) == 0:
return False, "请输入标签tags:list", None
rets = self.comm.Read(tags)
for ret in rets:
results.append(
{"tagName": ret.TagName,
"value": ret.Value,
"status": ret.Status
})
return True, "", results
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return False, str(io_error), None
connect_count += 1
self.connect()
self.get_tags(tags)
print(traceback.format_exc())
except Exception as e:
print(traceback.format_exc())
return False, str(e), None
def get_array(self, tag: str, count: int = 1):
"""
读取数组
:param tag: 目标标签
:param count: 读取个数
:return: [bool, str, (dict, None)]
"""
connect_count: int = 0
try:
if self.plcIp is None:
return False, "请输入PLC的IP地址", None
if not self.__isIP(self.plcIp):
return False, "PLC的IP地址格式异常", None
if len(tag) == 0:
return False, "请输入标签tags:list", None
if count <= 0:
return False, "取数个数应大于0", None
ret = self.comm.Read(tag, count)
results = {"tagName": ret.TagName,
"value": ret.Value,
"status": ret.Status
}
return True, "", results
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return False, str(io_error), None
connect_count += 1
self.connect()
self.get_array(tag, count)
except Exception as e:
print(traceback.format_exc())
return False, str(e), None
def get_tag_list(self):
"""
获取标签列表
:return:
"""
connect_count: int = 0
results = []
try:
tags = self.comm.GetTagList()
for t in tags.Value:
results.append({t.TagName: t.DataType})
return results
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return None
connect_count += 1
self.connect()
self.get_tag_list()
except Exception as e:
print(traceback.format_exc())
return None
def get_plc_time(self, raw=False):
"""
获取PLC时间
:param raw: 如果设置 raw=True，则将返回原始微秒，否则返回 [datetime] 2021-04-20 15:41:22.964380
:return:
"""
connect_count: int = 0
try:
time = self.comm.GetPLCTime(raw)
return time
except IOError as io_error:
if connect_count > 0:
print(traceback.format_exc())
return None
connect_count += 1
self.connect()
self.get_plc_time()
except Exception as e:
print(traceback.format_exc())
return None
# if __name__ == "__main__":
# plc = PLC_Tool("", 0)
#
# success, msg = plc.write_tag("PMCD", 0)

46
utils/LogT.py Normal file
View File

@ -0,0 +1,46 @@
import logging
from logging.handlers import RotatingFileHandler
import os
import sys
class Log(object):
def __init__(self, log_path="./logs", stdout_log_level=logging.DEBUG, fil_log_level=logging.WARN, log_name: str = "V"):
# self.log_level = stdout_log_level
self.logger = logging.getLogger(log_name)
self.logger.setLevel(stdout_log_level)
self.logPath = log_path
if not os.path.exists(self.logPath):
os.mkdir(self.logPath)
# 创建处理器
self.stdoutHandler = logging.StreamHandler(stream=sys.stdout)
self.fileHandler = RotatingFileHandler(filename="%s/log.log" % self.logPath,
mode="a",
maxBytes=10485760,
backupCount=10,
encoding="utf-8")
# 设置日志级别
self.stdoutHandler.setLevel(stdout_log_level)
self.fileHandler.setLevel(fil_log_level)
self.fmt = logging.Formatter(
"%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d >>> %(message)s"
)
# 设置日志格式
self.stdoutHandler.setFormatter(self.fmt)
self.fileHandler.setFormatter(self.fmt)
# 设置日志处理器
self.logger.addHandler(self.stdoutHandler)
self.logger.addHandler(self.fileHandler)
def getLogger(self):
return self.logger

56
utils/PlcTagsConfigT.py Normal file
View File

@ -0,0 +1,56 @@
import configparser
import json
from utils.Tool import logger, plc_tags_config_path
import re
plc_tags_config_ini = configparser.ConfigParser()
class PlcTagsConfigT:
def __init__(self):
# logger.info("PLC 标签读取地址：%s", plc_tags_config_path)
try:
plc_tags_config_ini.read(plc_tags_config_path, encoding="utf-8")
self.config_path = plc_tags_config_path
self.loading_1_tags_dict = {}
self.loading_2_tags_dict = {}
self.array_1_tags_dict = {}
self.array_2_tags_dict = {}
self.belt_tags_dict = {}
self.feeder_tags_dict = {}
self.loading_1_tags_list = []
self.loading_2_tags_list = []
self.array_1_tags_list = []
self.array_2_tags_list = []
self.belt_tags_list = []
self.feeder_tags_list = []
for section in plc_tags_config_ini.sections():
options = plc_tags_config_ini[section]
for option, value in options.items():
if section == "loading_1":
self.loading_1_tags_dict[option] = value
self.loading_1_tags_list.append([option, value])
elif section == "loading_2":
self.loading_2_tags_dict[option] = value
self.loading_2_tags_list.append([option, value])
elif section == "array_tags_1":
match = re.search(r'([A-Za-z._0-9]*)\((\d+)~(\d+)\)', value)
self.array_1_tags_list.append([option, *match.groups()])
elif section == "array_tags_2":
match = re.search(r'([A-Za-z._0-9]*)\((\d+)~(\d+)\)', value)
self.array_2_tags_list.append([option, *match.groups()])
elif section == "belt":
self.belt_tags_dict[option] = value
self.belt_tags_list.append([option, value])
elif section == "feeder":
self.feeder_tags_dict[option] = value
self.feeder_tags_list.append([option, value])
else:
logger.warn("读取到意外的配置项 section:%s", section)
except Exception as e:
logger.error("读取PLC点位配置文件失败，请检查配置文件位置和文件格式！")

38
utils/Tool.py Normal file
View File

@ -0,0 +1,38 @@
import traceback
from .ConnectPLC import PLC_Tool
from .ConfigT import ConfigT, configparser
from .LogT import logging, Log
logPath = "./logs"
config_path = "./config/config.ini"
plc_tags_config_path = "./config/plc_tags_config.ini"
config = None
try:
config = ConfigT(config_path)
except configparser.NoSectionError as e:
print("配置文件地址: %s \n %s" % (config_path, traceback.format_exc()))
def setLevel(log_level):
if log_level == "FATAL":
return logging.FATAL
elif log_level == "ERROR":
return logging.ERROR
elif log_level == "WARN":
return logging.WARN
elif log_level == "INFO":
return logging.INFO
else:
return logging.DEBUG
logger = Log(logPath, stdout_log_level=setLevel(config.stdout_log_level),
fil_log_level=setLevel(config.file_log_level)).getLogger()
logger.info("服务端口：%d 股道数量：%s", config.server_port, config.track_num)
logger.info("日志地址：%s 打印日志级别：%s 存储日志级别：%s", config.log_path, config.stdout_log_level,
config.file_log_level)
logger.info("PLC地址：%s 插槽：%d 超时：%d秒 最小读取间隔：%d毫秒", config.ip_address, config.slot, config.timeout,
config.interval_plc)
logger.debug("车厢最大数：%d", config.carriage_num_max)