-
Notifications
You must be signed in to change notification settings - Fork 0
/
state_machine.py
181 lines (143 loc) · 5.84 KB
/
state_machine.py
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
import datetime as dt
from enum import Enum
from logging import Logger, getLogger
from .event_model import ProgramInfo
LOGGER: Logger = getLogger(__package__ + ".state_machine")
class State(str, Enum):
"""State of media receiver."""
OFF = "off"
PLAYING = "playing"
PAUSED = "paused"
BUFFERING = "buffering"
class MediaReceiverStateMachine:
state: State | None = None
duration: int | None = None
position: int | None = None
position_last_update: dt.datetime | None
chan_key: int | None = None
program_current: ProgramInfo | None = None
program_next: ProgramInfo | None = None
_last_poll_player_state = None
_last_event_play_content = None
_last_event_eit_changed = None
_ignore_next_poll_event = False
_available: bool = False
def on_connection_error(self) -> None:
self._available = False
def on_event_eit_changed(self, data: dict) -> None:
LOGGER.debug("On Event EitChanged: %s", data)
self._available = True
if self._last_event_eit_changed == data:
LOGGER.debug("Event EitChanged is identical to last event. Ignoring")
return
self._on_event_eit_changed_changed(data)
self._last_event_eit_changed = data
def _on_event_eit_changed_changed(self, data) -> None:
if "channel_num" in data:
self.chan_key = int(data["channel_num"])
if "program_info" in data:
programm_info = data["program_info"]
self.program_current = ProgramInfo(**programm_info[0])
self.program_next = ProgramInfo(**programm_info[1])
def on_event_play_content(self, data: dict) -> None:
LOGGER.debug("On Event PlayContent: %s", data)
self._available = True
if self._last_event_play_content == data:
LOGGER.debug("Event PlayContent is identical to last event. Ignoring")
return
self._on_event_play_content_changed(data)
self._last_event_play_content = data
def _on_event_play_content_changed(self, data: dict) -> None:
if "new_play_mode" in data:
if data["new_play_mode"] == 20:
self.state = State.BUFFERING
self.duration = None
self.position = None
self.position_last_update = None
# poll api is always lagging behind. To prevent switching back and forth we ignore the next event and wait for a change
self._ignore_next_poll_event = True
return
# [2, 3, 4, 5]
elif data["new_play_mode"] == 4:
self.state = State.PLAYING
self.duration = 0
self.position = 0
self.position_last_update = dt.datetime.now()
# poll api is always lagging behind. To prevent switching back and forth we ignore the next event and wait for a change
self._ignore_next_poll_event = True
return
elif data["new_play_mode"] == 2:
# play after pause -> time shift ?
self.state = State.PLAYING
self.duration = data["duration"]
self.position = data["playPostion"]
self.position_last_update = dt.datetime.now()
return
elif data["new_play_mode"] == 1:
self.state = State.PAUSED
self.duration = data["duration"]
self.position = data["playPostion"]
self.position_last_update = dt.datetime.now()
return
elif data["new_play_mode"] == 0:
self.state = State.OFF
self._clear_non_state_attributes()
return
def on_poll_player_state(self, data: dict) -> None:
LOGGER.debug("On Poll PlayerState: %s", data)
self._available = True
if self._last_poll_player_state == data:
LOGGER.debug("Poll PlayerState is identical to last poll. Ignoring")
return
self._on_poll_player_state_changed(data)
self._last_poll_player_state = data
def _on_poll_player_state_changed(self, data: dict) -> None:
# deep sleep ?
if {"playBackState"} == set(data.keys()):
if data["playBackState"] == "0":
self.state = State.OFF
self._clear_non_state_attributes()
return
# tv running
if {
"chanKey",
"duration",
"mediaCode",
"mediaType",
"playBackState",
"playPostion",
} <= data.keys():
if "fastSpeed" in data and data["fastSpeed"] == "0":
self.state = State.PAUSED
else:
if self.state != State.PLAYING:
self.state = State.PLAYING
self.chan_key = int(data["chanKey"])
self.duration = int(data["duration"])
self.position = int(data["playPostion"])
self.position_last_update = dt.datetime.now()
return
if {
"chanKey",
"mediaCode",
"mediaType",
"playBackState",
} == set(data.keys()):
if self._last_poll_player_state is not None:
# this is an update and NOT the initial poll
if not self._ignore_next_poll_event:
self._ignore_next_poll_event = False
if self.state != State.OFF:
self.state = State.OFF
self._clear_non_state_attributes()
return
def _clear_non_state_attributes(self):
self.chan_key = None
self.duration = None
self.position = None
self.position_last_update = None
self.program_current = None
self.program_next = None
@property
def available(self) -> bool:
return self._available