MacMate/lid_manager.py at main · Jelahu/MacMate · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
#!/usr/bin/env python3
import sys
import time
import threading
import subprocess
import os

from pybooklid import LidSensor, is_sensor_available

AUDIO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "audio")
CLOSE_AUDIO_PATH = os.path.join(AUDIO_DIR, "no-no-wait-wait.mp3")
OPEN_AUDIO_PATH = os.path.join(AUDIO_DIR, "hello-there.mp3")

SAMPLE_INTERVAL = 0.03

# Closing trigger
OPEN_ANGLE_REQUIRED = 80.0
TRIGGER_CLOSE_BELOW_ANGLE = 60.0
MIN_CLOSING_SPEED = 45.0

# Opening trigger
CLOSED_ANGLE_REQUIRED = 25.0
TRIGGER_OPEN_ABOVE_ANGLE = 45.0
MIN_OPENING_SPEED = 45.0

MIN_ANGLE_CHANGE = 2.0
COOLDOWN = 0.1

_audio_lock = threading.Lock()
_current_audio_process = None


def play_audio_async(path: str):
    global _current_audio_process

    with _audio_lock:
        if _current_audio_process is not None:
            _current_audio_process.terminate()
            try:
                _current_audio_process.wait(timeout=0.2)
            except Exception:
                _current_audio_process.kill()

        _current_audio_process = subprocess.Popen(
            ["afplay", path],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )


def main():
    if not is_sensor_available():
        sys.exit("No lid angle sensor found.")

    previous_angle = None
    previous_time = None
    last_play = 0.0

    close_armed = False
    open_armed = False

    with LidSensor() as sensor:
        while True:
            angle = sensor.read_angle()
            now = time.time()

            if angle is None:
                time.sleep(SAMPLE_INTERVAL)
                continue

            if angle >= OPEN_ANGLE_REQUIRED:
                close_armed = True

            if angle <= CLOSED_ANGLE_REQUIRED:
                open_armed = True

            if previous_angle is None:
                previous_angle = angle
                previous_time = now
                time.sleep(SAMPLE_INTERVAL)
                continue

            dt = max(now - previous_time, 1e-6)
            angle_change = angle - previous_angle
            speed = angle_change / dt

            previous_angle = angle
            previous_time = now

            closing_speed = -speed
            opening_speed = speed

            is_closing = (
                close_armed
                and angle <= TRIGGER_CLOSE_BELOW_ANGLE
                and angle_change <= -MIN_ANGLE_CHANGE
                and closing_speed >= MIN_CLOSING_SPEED
            )

            is_opening = (
                open_armed
                and angle >= TRIGGER_OPEN_ABOVE_ANGLE
                and angle_change >= MIN_ANGLE_CHANGE
                and opening_speed >= MIN_OPENING_SPEED
            )

            can_play = (now - last_play) >= COOLDOWN

            if is_closing and can_play:
                play_audio_async(CLOSE_AUDIO_PATH)
                last_play = now
                close_armed = False

            elif is_opening and can_play:
                play_audio_async(OPEN_AUDIO_PATH)
                last_play = now
                open_armed = False

            time.sleep(SAMPLE_INTERVAL)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        with _audio_lock:
            if _current_audio_process is not None:
                _current_audio_process.terminate()