Timer

class hive.trigger.Timer(**event_config)

Bases: QThread, Trigger

Timer Trigger.

Use in the Config file

Task:
  mode: ...
  trigger:
    - name: Timer
      options:
        event_name_1: # routine timer
          interval: 1 # 1 seconds           # (float) time interval.
          event_type: "routine"             # (str) 'routine', 'one_shot', 'continue'.
        event_name_2: # one shot timer
          interval: 600 # 10 minutes
          event_type: "one_shot"
        event_name_3: # continue timer
          interval: 1
          event_type: "continue"
          repeat_times: 10                 # (int) repeat firing times. None for infinite times. Only used in 'continue' timer.

Use in task

Code implementation:

# get timer trigger from somewhere
timer: Timer

# create a timer handle
h = timer.add_timer('event_name_1', 60, "routine")

# start a timer handle
h.start()

Read the event in the task:

class MyTask(Task):
    def filter_event(self, event_type: str, event_data: str):
        if event_type == "timer" and event_data == "event_name_1":
            ...
        elif event_type == "timer" and event_data == "event_name_2":
            ...

Event Type

• routine: routine firing timer event regularly (fixed interval).

• one_shot: firing timer event once.

• continue: continuous firing timer event (vary interval).

Attributes Summary

INSTANCE
TYPE_CONTINUE
TYPE_ONE_SHOT
TYPE_ROUTINE
timer_event

Methods Summary

add_timer(event_name[, interval, ...])	create a timer handle.
cancel_all()	cancel all running timers.
clear_all()	cancel all running timers and remove all.
run(self)
who_to_connect_where()

Attributes Documentation

INSTANCE: ClassVar[Timer] = None

TYPE_CONTINUE = 'continue'

TYPE_ONE_SHOT = 'one_shot'

TYPE_ROUTINE = 'routine'

timer_event

Methods Documentation

add_timer(event_name: Optional[str], interval: float = 1, event_type: str = 'one_shot', repeat_times: Optional[int] = None, callback: Optional[Callable[[str, int], Optional[bool]]] = None, replace=False) → TimerHandle

create a timer handle.

Parameters

• event_name (str) – event name

• interval (float) – event interval. unit second.

• event_type ({'routine', 'one_shot', 'continue'}) – event type

• repeat_times – repeat times when event_type == ‘continue’. None for infinite.

• callback – timer callback with signature (event_name:str, counter:int) -> continue:bool?. If None, emit Timer.timer_event(f”timer:{event_name}”).

• replace – replace existed timer handle.

Returns

timer handle

Return type

TimerHandle

cancel_all()

cancel all running timers.

clear_all()

cancel all running timers and remove all.

run(self)

who_to_connect_where()