此脚本功能为根据工作日开关PLC控制器大屏,工作日包含自动检测单双休与法定节假日调休。
使用 python 完成逻辑,mysql 存储日期以及单双休。
| 名称 | 类型 | 备注 |
|---|---|---|
| day | date | 日期 唯一 主键 |
| status | int | 1 周末 2 法定节假日 3 自定义 |
| dayoff | int | 是否休息 1休息 0为不休或调休 |
| info | varchar | 备注 |
脚本使用 Python 实现,使用中国节日接口拉取调休和周末信息。存储后标记日期是否休息。每天执行脚本时查表,如日期不在表内或者是被标记为调休、大小休的上班日,则不开启屏幕。否则在早上八点钟通过 TCP 信号开启大屏幕。
pythonOPEN = 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)
pythondef 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
pythonimport 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()