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

目录

功能
结构
脚本部分
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()