smart_alarm_clock.py

from typing import Optional, Deque

import sys
from alarm import BaseAlarmClock, AlarmState
from datetime import time, datetime, timedelta, timezone
from time import sleep
# from pynput import keyboard
import keyboard as kb
from collections import deque
import numpy as np
# import tensorflow as tf
import tflite_runtime.interpreter as tflite
import vlc

# Lower threshold means it only needs to be somewhat probable to be early wake up
# >= 1 means never allow early wake up in that time
WAKE_THRESHOLD = np.array([0.5, 0.75, 0.9, 1.0])

# SNOOZE_KEY = keyboard.Key.up
SNOOZE_KEY = "space"
# ALARM_OFF_KEY = keyboard.Key.esc
ALARM_OFF_KEY = "esc"
MODEL_MEMORY: int = 5

SECONDS_IN_MINUTE: float = 60
SNOOZE_SEC: float = 60
WAIT_SEC: float = 1

MODEL_PATH: str = ".model/attention_model.tflite"
VLC_INSTANCE = vlc.Instance("--input-repeat=999")
VLC_PLAYER = VLC_INSTANCE.media_player_new()
SONG = VLC_INSTANCE.media_new(".model/Viennese Poets.mp3" )

# Input Array: [minutes_since_start, current_hour_utc, current_minute_utc, awake_prob, rem_prob, light_prob, deep_prob]
# Output Array: [awake_prob, rem_prob, light_prob, deep_prob]


def softmax(arr):
    val = np.exp(arr)
    return val / sum(val)


def model_prediction(model_func, current_times_prob: Deque[np.array]) -> np.array:
    # print(f"Prob shape: {np.array([current_times_prob]).shape}")
    if model_func:
        # prediction = model.predict(np.array([current_times_prob]))[0]
        prediction = model_func(x=np.array([current_times_prob]))
        print(prediction)
        # print(f"prediction: {prediction}")
        # print(f"softmax: {softmax(prediction)}")
        return softmax(prediction)
    else:
        return np.zeros(4)


class SmartAlarmClock(BaseAlarmClock):

    def __init__(self,
                 wake_time: Optional[datetime] = None,
                 earliest_wake: Optional[timedelta] = None,
                 wake_threshold: np.array = WAKE_THRESHOLD,
                 vlc_player=VLC_PLAYER,
                 default_song=SONG):
        super().__init__(wake_time)
        self.earliest_wake = self.wake_time
        if wake_time:
            self.set_alarm(wake_time, earliest_wake)
        self.wake_threshold = wake_threshold
        self._vlc_player = vlc_player
        self.song = default_song
        if self._vlc_player and self.song:
            self._vlc_player.set_media(self.song)

    def set_alarm(self, wake_time: datetime, earliest_wake: Optional[timedelta] = None) -> datetime:
        print(f"SETTING ALARM to: {wake_time}")
        returned_value = super(SmartAlarmClock, self).set_alarm(wake_time)
        self.earliest_wake = wake_time - earliest_wake if earliest_wake else self.wake_time
        print(returned_value)
        return returned_value

    def start_alarm(self) -> None:
        print("STARTING ALARM")
        super(SmartAlarmClock, self).start_alarm()

    def sound_alarm(self, override_song=None) -> None:
        print("SOUNDING ALARM")
        if override_song:
            self.song = override_song
        if self._vlc_player:
            self._vlc_player.set_media(self.song)
            self._vlc_player.play()
        super(SmartAlarmClock, self).sound_alarm()

    def snooze_alarm(self) -> None:
        print("SNOOZING ALARM")
        super(SmartAlarmClock, self).snooze_alarm()
        if self._vlc_player:
            self._vlc_player.pause()
        sleep(SNOOZE_SEC)
        if self._vlc_player:
            self._vlc_player.play()
        print("UN-SNOOZING ALARM")
        self.sound_alarm()

    def stop_alarm(self, deactivate: bool = True) -> None:
        print("STOPPING ALARM")
        if self._vlc_player:
            self._vlc_player.stop()
            print("STOPPED PLAYING ALARM")
        super(SmartAlarmClock, self).stop_alarm()

    def check_early_alarm_reached(self, current_time: datetime, prior_prediction: np.array) -> bool:
        if current_time < self.earliest_wake:
            return False
        return np.max(prior_prediction - self.wake_threshold) > 0

    def smart_alarm_mode(self, fn, model_memory=MODEL_MEMORY):
        print("STARTING IN SMART MODE")
        kb.on_press_key("c", lambda x: sys.exit())
        kb.on_press_key(SNOOZE_KEY, lambda x: self.snooze_alarm())
        kb.on_press_key(ALARM_OFF_KEY, lambda x: self.stop_alarm())
        self.start_alarm()
        time_queue = deque(maxlen=model_memory)
        # While not filled time_queue keep appending awake data (warm-up machine)
        minutes_since_start = 0
        while len(time_queue) < time_queue.maxlen:
            current_utc = datetime.utcnow()
            time_queue.append(np.array(
                [minutes_since_start, current_utc.hour, current_utc.minute, 1.0, 0.0, 0.0, 0.0]
            ))
            print(f"time_queue: {time_queue}")
            # Wait a minute for next timestep
            while (datetime.utcnow() - current_utc).seconds < SECONDS_IN_MINUTE:
                sleep(WAIT_SEC)
            minutes_since_start += 1

        # While the alarm clock is running
        last_checked = datetime.utcnow()
        last_prediction = model_prediction(fn, time_queue)
        print(f"{last_checked}: last_prediction = {last_prediction}")
        print(f"check early: {self.check_early_alarm_reached(datetime.now(tz=timezone.utc), last_prediction)}")
        while self.current_state is AlarmState.RUNNING \
                and not self.check_early_alarm_reached(datetime.now(tz=timezone.utc), last_prediction) \
                and not self.alarm_check_reached(datetime.now(tz=timezone.utc)):
            print(f"ALARM SLEEPING @: {datetime.now()}")
            if (datetime.utcnow() - last_checked).seconds < SECONDS_IN_MINUTE:
                last_checked = datetime.utcnow()
                time_queue.append(
                    np.concatenate(
                        (
                            [minutes_since_start, last_checked.hour, last_checked.minute],
                            last_prediction
                        )
                    )
                )
                print(f"time_queue = {time_queue}")
            last_prediction = model_prediction(fn, time_queue)
            print(f"{datetime.utcnow()}: last_prediction = {last_prediction}")
            sleep(WAIT_SEC)

        # While the alarm clock is sounding off
        self.sound_alarm()
        while self.current_state is not AlarmState.DEACTIVATED:
            pass
            # event = kb.read_event()
            # if event.event_type == kb.KEY_DOWN:
            # # with keyboard.Events() as events:
            # #     for event in events:
            # #         if event.key == ALARM_OFF_KEY:
            #         if event.name == ALARM_OFF_KEY:
            #             self.stop_alarm()
            #             break
            #         elif event.name == SNOOZE_KEY:
            #         # elif event.key == SNOOZE_KEY:
            #             self.snooze_alarm()
            #             break


if __name__ == '__main__':
    alarm_hour = int(input("What hour do you want the alarm to go off at? "))
    alarm_minute = int(input("What minute do you want the alarm to go off at? "))
    early_minutes = int(input("How many minute early would you have the alarm go off at? "))
    run_alarm_today = map(
        lambda response: response.lower() in ["y", "yes", "t", "true"],
        input("Will the alarm run today? [y/N]")
    )
    alarm_time = datetime.combine(
        datetime.today() if run_alarm_today else (datetime.today() + timedelta(days=1)).date(),
        time(alarm_hour, alarm_minute, 0)
    ).astimezone(tz=timezone.utc)
    early_time = alarm_time - timedelta(minutes=early_minutes)
    alarm_clock_class = SmartAlarmClock(alarm_time, earliest_wake=timedelta(minutes=early_minutes))
    print(f"Loading model from: {MODEL_PATH}")
    ai_model = tflite.Interpreter(model_path=MODEL_PATH)
    ai_model.allocate_tensors()  # tf.keras.models.load_model(MODEL_PATH)
    model_fn = ai_model.get_signature_runner()
    print(f"Initialized Model")
    alarm_clock_class.smart_alarm_mode(model_fn)