LAS/core/time_plan.py

from typing import List, Tuple, Optional
from datetime import datetime, time, date
from threading import Thread, Lock
from enum import Enum

from common.sys_comm import (
    LOGDBG, LOGINFO, LOGWARN, LOGERR,
    get_utc_time_ms
)


# 时间计划
class TimePlan:
    def __init__(self,
                 time_range: list,
                 start_date: str,
                 stop_date: str,
                 weekdays: Optional[List[int]] = None,
                 month_days: Optional[List[int]] = None
                 ):

        self.start_date_    = start_date    # 开始日期
        self.stop_date_     = stop_date     # 结束日期
        self.time_range_    = time_range    # 生效时间
        # [{"start_time": "00:00","end_time": "12:00"},{"start_time": "18:00","end_time": "23:59"}, ...]
        self.month_days_    = month_days    # 每月的生效日期
        self.weekdays_      = weekdays      # 每周的生效日期

    def is_active_now(self, now: Optional[datetime] = None) -> bool:
        now = now or datetime.now()

        # 起止日期匹配
        try:
            if not (self.start_date_ <= now.strftime("%Y-%m-%d") <= self.stop_date_):
                return False
        except ValueError:
            return False

        # 周日期匹配
        if self.weekdays_ is not None and now.weekday() not in self.weekdays_:
            return False

        # 月日期匹配
        if self.month_days_ is not None and now.day not in self.month_days_:
            return False


        # 时间匹配
        now_t = now.time()
        for period in self.time_range_:
            start_t = datetime.strptime(period["start_time"], "%H:%M").time()
            end_t = datetime.strptime(period["end_time"], "%H:%M").time()
            if start_t <= now_t <= end_t:
                return True

        return False

    def example(self) -> bool:
        if self.is_active_now():
            return True
        else:
            return False