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
|
import logging
import PyATEMMax
from PyATEMMax.ATEMProtocolEnums import ATEMVideoModeFormats, ATEMTransitionStyles
from rich import inspect
VIDEO_FORMATS = {
f[1:]
for f in dir(ATEMVideoModeFormats)
if f.startswith('f')
}
class PyATEMSwitcher:
def __init__(self, config):
self.atem = PyATEMMax.ATEMMax()
self.config = config.get('atem', {})
self.log = logging.getLogger('Switcher')
self._connect_subscribers = []
self._connect_attempt_subscribers = []
self._disconnect_subscribers = []
self._receive_subscribers = []
self._validate_config()
self.atem.registerEvent(
self.atem.atem.events.connect,
self._on_connect,
)
self.atem.registerEvent(
self.atem.atem.events.connectAttempt,
self._on_connect_attempt,
)
self.atem.registerEvent(
self.atem.atem.events.disconnect,
self._on_disconnect,
)
self.atem.registerEvent(
self.atem.atem.events.receive,
self._on_receive,
)
def _on_connect(self, params):
self.log.debug(f'_on_connect({repr(params)})')
self._push_config()
for callback in self._connect_subscribers:
callback(params)
def _on_connect_attempt(self, params):
self.log.debug(f'_on_connect_attempt({repr(params)})')
for callback in self._connect_attempt_subscribers:
callback(params)
def _on_disconnect(self, params):
self.log.debug(f'_on_disconnect({repr(params)})')
for callback in self._disconnect_subscribers:
callback(params)
def _on_receive(self, params):
self.log.debug(f'_on_receive({repr(params)})')
for callback in self._receive_subscribers:
callback(params)
def _push_config(self):
conf = self.config.get('settings', {})
# TODO media upload to MP1
if 'video_mode' in self.config:
video_mode = getattr(
ATEMVideoModeFormats,
'f'+self.config['video_mode'],
)
if self.atem.videoMode.format != video_mode:
self.atem.setVideoModeFormat(video_mode)
if conf.get('inputs', None):
for inp in self.atem.atem.videoSources:
if conf['inputs'].get(inp.value, None):
new_name = conf['inputs'][inp.value]
self.log.debug(f"setting input {inp.value} to name '{new_name}'")
self.atem.setInputLongName(inp.value, new_name)
self.atem.setInputShortName(inp.value, new_name[0:3])
def _validate_config(self):
if 'ip' not in self.config:
raise KeyError('Please set ATEM IP in config!')
if (
'video_mode' in self.config
and self.config['video_mode'] not in VIDEO_FORMATS
):
raise ValueError(
f'ATEM video_mode {self.config["video_mode"]} '
'is not a valid video mode, must be one of: '
f'{", ".join(sorted(VIDEO_FORMATS))}'
)
def connect(self):
self.log.info('Initiating connection to switcher')
self.atem.connect(self.config['ip'])
def disconnect(self):
self.atem.disconnect()
def on_connect(self, callback):
self._connect_subscribers.append(callback)
def on_connect_attempt(self, callback):
self._connect_attempt_subscribers.append(callback)
def on_disconnect(self, callback):
self._disconnect_subscribers.append(callback)
def on_receive(self, callback):
self._receive_subscribers.append(callback)
def trans(self, input):
self.log.debug(f'hehehehe trans({repr(input)})')
self.atem.setPreviewInputVideoSource(
ATEMTransitionStyles.mix,
input,
)
self.atem.setTransitionMixRate(0, 10)
self.atem.execAutoME(0)
|
