编辑
2025-07-10
运维
00
请注意,本文编写于 381 天前,最后修改于 0 天前,其中某些信息可能已经过时。

目录

功能
结构
脚本部分
TCP 信号开启 plc 电源
开启大屏逻辑
 定时
手动控制
屏幕控制
日期控制
PLC电源脚本

功能

此脚本功能为根据工作日开关PLC控制器大屏,工作日包含自动检测单双休与法定节假日调休。

使用 python 完成逻辑,mysql 存储日期以及单双休。

结构

名称类型备注
daydate日期 唯一 主键
statusint1 周末
2 法定节假日
3 自定义
dayoffint是否休息 1休息 0为不休或调休
infovarchar备注

脚本部分

脚本使用 Python 实现,使用中国节日接口拉取调休和周末信息。存储后标记日期是否休息。每天执行脚本时查表,如日期不在表内或者是被标记为调休、大小休的上班日,则不开启屏幕。否则在早上八点钟通过 TCP 信号开启大屏幕。

TCP 信号开启 plc 电源

python
OPEN = bytes.fromhex('000000000006000613E80001') CLOSE = bytes.fromhex('000000000006000613E90001') # host为目标ip地址,comm为上面十六进制的OPEN或者CLOSE信号 def TcpControl(host,comm,port=5000): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) # 连接到目标 s.sendall(comm)

开启大屏逻辑

python
def goOpen(host): """ 开启host指向的plc电源 """ comm = OPEN TcpControl(host,comm) print(host,"开启") def openJob(): """ 查表检查是否是工作日,非工作日直接return """ time = datetime.datetime.now().strftime('%Y-%m-%d') config = initConfig() selectSql = "SELECT dayoff FROM `tool`.`date` WHERE `day` = %s " with DB(**config) as cur: cur.execute(selectSql,time) res = cur.fetchone() if res and res["dayoff"]: print("非工作日不需要开启") return goOpen("ip")

 定时

定时通过 Linux 自带的 Cron 配置实现

shell
#打开cron配置文件 crontab -e # 逻辑实现如下 0 8 * * * python3 /root/plcPower/plcPower.py open # 每天早上八点检测开启大屏 0 17 * * * python3 /root/plcPower/plcPower.py close # 每天下午五点关闭大屏

手动控制

屏幕控制

同时手动执行脚本也可起到一样的作用,具体命令如下

shell
# 关闭大屏 python3 /root/plcPower/plcPower.py close # 开启大屏 python3 /root/plcPower/plcPower.py open # 大小休反转 python3 /root/plcPower/plcPower.py rever

日期控制

如需要手动添加不需要开启屏幕的日期,可查看日期是否在表中

如已经在表中,可以将 dayoff 字段改为 0 则当天不会开启大屏幕

如未在表中则添加日期, status 为 3 ,dayoff 为 0

PLC电源脚本

python
import socket import requests import datetime import sys from time import sleep from mysqlDB import DB from config import Config OPEN = bytes.fromhex('000000000006000613E80001') CLOSE = bytes.fromhex('000000000006000613E90001') def initConfig(): congif,logger = Config().getConfig("mysql") config = dict(congif) config["port"] = int(config["port"]) return config def TcpControl(host,comm,port=5000): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) # 连接到目标 s.sendall(comm) def getDate(): """ 通过接口获取节假日和调休 返回(xxxx-xx-xx,假日类型,是否休息,备注)的list """ url1 = "https://api.jiejiariapi.com/v1/holidays/2025" url2 = "https://api.jiejiariapi.com/v1/weekends/2025" insertList = [] resp = requests.get(url=url1) holiday = resp.json() for date in holiday: insertList.append((holiday[date]["date"],1,1 if holiday[date]["isOffDay"] else 0,holiday[date]["name"])) listlen = len(insertList) print("holidays insert success") resp = requests.get(url=url2) dayoff = resp.json() for num,date in enumerate(dayoff): f = 0 for index in range(listlen): if insertList[index][0] == dayoff[date]["date"]: f = 1 if f : continue if dayoff[date]["name"] == "周日": insertList.append((dayoff[date]["date"],2,1,dayoff[date]["name"])) continue if num/2%2 == 1: insertList.append((dayoff[date]["date"],2,1,dayoff[date]["name"])) continue insertList.append((dayoff[date]["date"],2,0,dayoff[date]["name"])) print("offday insert success") return insertList def initDate(): """ 写入数据库 """ config = initConfig() insertSql = """ INSERT INTO `tool`.`date` (`day`, `status`, `dayoff`, `info`) VALUES (%s, %s, %s, %s) """ value = getDate() with DB(**config) as cur: cur.executemany(insertSql,value) def goClose(host): comm = CLOSE TcpControl(host,comm) print(host,"关闭") def goOpen(host): comm = OPEN TcpControl(host,comm) print(host,"开启") def openJob(): time = datetime.datetime.now().strftime('%Y-%m-%d') config = initConfig() selectSql = "SELECT dayoff FROM `tool`.`date` WHERE `day` = %s " with DB(**config) as cur: cur.execute(selectSql,time) res = cur.fetchone() if res and res["dayoff"]: print("非工作日不需要开启") return goOpen("192.168.132.103") def rever(): """ 大小休反转 """ config = initConfig() selSql = "SELECT * FROM `tool`.`date` WHERE `info` = '周六' AND `status` = '2'" updateSql = "UPDATE `tool`.`date` SET `dayoff` = %s WHERE `day` = %s" value = [] with DB(**config) as cur: cur.execute(selSql) res = cur.fetchall() # print(res) for index in res: if index["dayoff"]: index["dayoff"] = 0 continue index["dayoff"] = 1 # print(res) for index in res: value.append((index["dayoff"],index["day"])) print(value) with DB(**config) as cur: cur.executemany(updateSql,value) def closeJob(): time = datetime.datetime.now().strftime('%Y-%m-%d') print("定时关闭",time) goClose("192.168.132.103") def main(): # schedule.every().day.at("08:00").do(openJob) # schedule.every().day.at("11:55").do(closeJob) # while True: # schedule.run_pending() if len(sys.argv) < 2 or len(sys.argv) > 2: print("请提供启动参数\n示例: open close rever\n 此程序用于关闭和开启宏图阁大厅LED的PLC电源 \n rever可以用于大小休反转") sys.exit(1) if "open" in sys.argv[1]: openJob() if "close" in sys.argv[1]: closeJob() if "rever" in sys.argv[1]: rever() # sleep(20) # closeJob() if __name__ == '__main__': main() # rever()