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
|
"""Snapcast client."""
import logging
_LOGGER = logging.getLogger(__name__)
# pylint: disable=too-many-public-methods
class Snapclient():
"""Represents a snapclient."""
def __init__(self, server, data):
"""Initialize."""
self._server = server
self._snapshot = None
self._last_seen = None
self._callback_func = None
self.update(data)
def update(self, data):
"""Update client."""
self._client = data
@property
def identifier(self):
"""Get identifier."""
return self._client.get('id')
@property
def group(self):
"""Get group."""
for group in self._server.groups:
if self.identifier in group.clients:
return group
return None
@property
def friendly_name(self):
"""Get friendly name."""
if len(self._client.get('config').get('name')):
return self._client.get('config').get('name')
return self._client.get('host').get('name')
@property
def version(self):
"""Version."""
return self._client.get('snapclient').get('version')
@property
def connected(self):
"""Connected or not."""
return self._client.get('connected')
@property
def name(self):
"""Name."""
return self._client.get('config').get('name')
async def set_name(self, name):
"""Set a client name."""
if not name:
name = ''
self._client['config']['name'] = name
await self._server.client_name(self.identifier, name)
@property
def latency(self):
"""Latency."""
return self._client.get('config').get('latency')
async def set_latency(self, latency):
"""Set client latency."""
self._client['config']['latency'] = latency
await self._server.client_latency(self.identifier, latency)
@property
def muted(self):
"""Muted or not."""
return self._client.get('config').get('volume').get('muted')
async def set_muted(self, status):
"""Set client mute status."""
new_volume = self._client['config']['volume']
new_volume['muted'] = status
self._client['config']['volume']['muted'] = status
await self._server.client_volume(self.identifier, new_volume)
_LOGGER.debug('set muted to %s on %s', status, self.friendly_name)
@property
def volume(self):
"""Volume percent."""
return self._client.get('config').get('volume').get('percent')
async def set_volume(self, percent, update_group=True):
"""Set client volume percent."""
if percent not in range(0, 101):
raise ValueError('Volume percent out of range')
new_volume = self._client['config']['volume']
new_volume['percent'] = percent
self._client['config']['volume']['percent'] = percent
await self._server.client_volume(self.identifier, new_volume)
if update_group:
self._server.group(self.group.identifier).callback()
_LOGGER.debug('set volume to %s on %s', percent, self.friendly_name)
def groups_available(self):
"""Get available group objects."""
return list(self._server.groups)
def update_volume(self, data):
"""Update volume."""
self._client['config']['volume'] = data['volume']
_LOGGER.debug('updated volume on %s', self.friendly_name)
self._server.group(self.group.identifier).callback()
self.callback()
def update_name(self, data):
"""Update name."""
self._client['config']['name'] = data['name']
_LOGGER.debug('updated name on %s', self.friendly_name)
self.callback()
def update_latency(self, data):
"""Update latency."""
self._client['config']['latency'] = data['latency']
_LOGGER.debug('updated latency on %s', self.friendly_name)
self.callback()
def update_connected(self, status):
"""Update connected."""
self._client['connected'] = status
_LOGGER.debug('updated connected status to %s on %s', status, self.friendly_name)
self.callback()
def snapshot(self):
"""Snapshot current state."""
self._snapshot = {
'name': self.name,
'volume': self.volume,
'muted': self.muted,
'latency': self.latency
}
_LOGGER.debug('took snapshot of current state of %s', self.friendly_name)
async def restore(self):
"""Restore snapshotted state."""
if not self._snapshot:
return
await self.set_name(self._snapshot['name'])
await self.set_volume(self._snapshot['volume'])
await self.set_muted(self._snapshot['muted'])
await self.set_latency(self._snapshot['latency'])
self.callback()
_LOGGER.debug('restored snapshot of state of %s', self.friendly_name)
def callback(self):
"""Run callback."""
if self._callback_func and callable(self._callback_func):
self._callback_func(self)
def set_callback(self, func):
"""Set callback function."""
self._callback_func = func
def __repr__(self):
"""Return string representation."""
return f'Snapclient {self.version} ({self.friendly_name}, {self.identifier})'
|
