193 lines
6.3 KiB
Python
193 lines
6.3 KiB
Python
# !/usr/bin/python3
|
||
"""
|
||
# @Time : 2025/12/12 11:05
|
||
# @Author : reenrr
|
||
# @File : util_time.py
|
||
# @Desc :
|
||
"""
|
||
import time
|
||
|
||
class MyTimer:
|
||
@staticmethod
|
||
def gMyGetTickCount():
|
||
"""
|
||
获取当前系统时间的毫秒级时间戳
|
||
"""
|
||
ts = time.time()
|
||
return int(ts * 1000) # Convert to milliseconds
|
||
|
||
# 通断延时器类
|
||
class CTon:
|
||
def __init__(self):
|
||
self.m_unET = 0
|
||
self.m_bLastI = False
|
||
self.m_bIn = False
|
||
self.m_bPause = False
|
||
self.m_bOver = True
|
||
self.m_unPT = 0
|
||
self.m_unStartTime = 0
|
||
self.m_unPauseET = 0
|
||
|
||
def GetET(self):
|
||
"""
|
||
获取当前已流逝的延时时间
|
||
"""
|
||
if self.m_bIn:
|
||
nET = self.m_unPT + (self.m_unStartTime - MyTimer.gMyGetTickCount())
|
||
return max(nET, 0)
|
||
else:
|
||
return 0
|
||
|
||
def SetReset(self):
|
||
"""
|
||
复位TON所有状态属性，终止当前延时流程，恢复到初始状态，为下一次延时流程做准备
|
||
"""
|
||
self.m_bIn = False
|
||
self.m_bLastI = False
|
||
self.m_bPause = False
|
||
|
||
def SetPause(self, value):
|
||
"""
|
||
设置延时暂停/恢复状态，实现延时过程的暂停与继续
|
||
:param value: 暂停标志 True表示暂停，False表示恢复
|
||
"""
|
||
if self.m_bIn: # 仅当输入有效时，运行设置暂停
|
||
self.m_bPause = value
|
||
if self.m_bPause: # 暂停时，记录当前已流逝的延时时间，用于恢复记时补偿
|
||
self.m_unPauseET = MyTimer.gMyGetTickCount() - self.m_unStartTime
|
||
|
||
def SetOver(self, value):
|
||
"""
|
||
手动设置延时完成标记
|
||
"""
|
||
self.m_bOver = value
|
||
|
||
def GetStartTime(self):
|
||
"""
|
||
获取延时启动时的时间戳
|
||
"""
|
||
return self.m_unStartTime
|
||
|
||
def Q(self, value_i, value_pt):
|
||
"""
|
||
触发延时逻辑，判断延时是否完成
|
||
:param value_i: 当前输入信号，True表示输入有效，False表示无效
|
||
:param value_pt: 延时时间，单位毫秒
|
||
:return: True表示延时完成，False表示延时未完成
|
||
"""
|
||
self.m_bIn = value_i
|
||
self.m_unPT = value_pt
|
||
un_tick = MyTimer.gMyGetTickCount()
|
||
|
||
if self.m_bOver and self.m_bIn:
|
||
self.m_unStartTime = un_tick - self.m_unPT
|
||
self.m_bOver = False
|
||
|
||
if self.m_bPause and self.m_bIn:
|
||
self.m_unStartTime = un_tick - self.m_unPauseET
|
||
|
||
if self.m_bIn != self.m_bLastI:
|
||
self.m_bLastI = self.m_bIn
|
||
if self.m_bIn:
|
||
self.m_unStartTime = un_tick
|
||
self.m_bPause = False
|
||
|
||
return self.m_bIn and (un_tick >= (self.m_unStartTime + self.m_unPT))
|
||
|
||
# 时钟脉冲类
|
||
class CClockPulse:
|
||
def __init__(self):
|
||
self.m_bFirstOut = True # 首次输出标记
|
||
self.m_bTonAOut = False # 延时器A的输出状态
|
||
self.m_bTonBOut = False # 延时器B的输出状态
|
||
self.m_cTonA = CTon() # 延时器A
|
||
self.m_cTonB = CTon() # 延时器B
|
||
|
||
def Q(self, value_i, run_time, stop_time):
|
||
"""
|
||
生成周期性脉冲信号，返回当前脉冲输出状态
|
||
:param value_i: 输入信号，True表示输入有效，False表示无效
|
||
:param run_time: 运行时间，单位毫秒
|
||
:param stop_time: 停止时间，单位毫秒
|
||
:return: True表示脉冲输出有效，False表示脉冲输出无效
|
||
"""
|
||
if self.m_bFirstOut: # 首次输出：按照[运行时间->停止时间]的顺序生成脉冲
|
||
self.m_bTonAOut = self.m_cTonA.Q(not self.m_bTonBOut and value_i, run_time)
|
||
self.m_bTonBOut = self.m_cTonB.Q(self.m_bTonAOut and value_i, stop_time)
|
||
return not self.m_bTonAOut and value_i
|
||
else: # 非首次输出：循环交替[停止时间->运行时间]，生成周期性脉冲
|
||
self.m_bTonAOut = self.m_cTonA.Q(not self.m_bTonBOut and value_i, stop_time)
|
||
self.m_bTonBOut = self.m_cTonB.Q(self.m_bTonAOut and value_i, run_time)
|
||
return self.m_bTonAOut and value_i
|
||
|
||
# 延时输出类，显示先等待，后输出，输出完成后自动复位的逻辑
|
||
class CDelayOut:
|
||
def __init__(self):
|
||
self.m_cOutTon = CTon()
|
||
self.m_cmWaitTon = CTon()
|
||
|
||
def Reset(self):
|
||
"""
|
||
复位两个延时器的所有状态
|
||
"""
|
||
self.m_cOutTon.SetReset()
|
||
self.m_cmWaitTon.SetReset()
|
||
|
||
def Q(self, value_i, wait_time, out_time):
|
||
"""
|
||
实现等待->输出->复位的闭环逻辑
|
||
:param value_i: 输入信号，True表示输入有效，False表示无效
|
||
:param wait_time: 等待时间，单位毫秒
|
||
:param out_time: 输出时间，单位毫秒
|
||
"""
|
||
if self.m_cmWaitTon.Q(value_i, wait_time):
|
||
if self.m_cOutTon.Q(True, out_time):
|
||
self.m_cOutTon.SetReset()
|
||
self.m_cmWaitTon.SetReset()
|
||
value_i = False
|
||
return False
|
||
return True
|
||
return False
|
||
|
||
# 边沿检测类，上升沿或下降沿触发
|
||
class CRisOrFall:
|
||
def __init__(self):
|
||
self.m_bTemp = False
|
||
|
||
def Q(self, value_i, ris_or_fall):
|
||
"""
|
||
边沿检测方法：检测输入信号的上升沿或下降沿，返回检测结果
|
||
"""
|
||
result = False
|
||
if value_i != self.m_bTemp:
|
||
if ris_or_fall and value_i: # Rising edge
|
||
result = True
|
||
if not ris_or_fall and not value_i: # Falling edge
|
||
result = True
|
||
self.m_bTemp = value_i
|
||
return result
|
||
|
||
# 断通电延时器类
|
||
class CTof:
|
||
def __init__(self):
|
||
self.m_cDelayTon = CTon()
|
||
self.m_bValue = False
|
||
self.m_cRis = CRisOrFall()
|
||
|
||
def SetReset(self):
|
||
self.m_bValue = False
|
||
self.m_cDelayTon.SetReset()
|
||
|
||
def Q(self, value_i, delay_time):
|
||
if self.m_cRis.Q(value_i, False):
|
||
self.m_cDelayTon.SetReset()
|
||
self.m_bValue = True
|
||
if self.m_cDelayTon.Q(self.m_bValue, delay_time):
|
||
self.m_bValue = False
|
||
self.m_cDelayTon.SetReset()
|
||
return value_i or self.m_bValue
|
||
|
||
# Utility function
|
||
def gGetNowTime():
|
||
return int(time.time())
|