-
-
Notifications
You must be signed in to change notification settings - Fork 30
Expand file tree
/
Copy pathctrl.py
More file actions
1481 lines (1323 loc) Β· 60.9 KB
/
ctrl.py
File metadata and controls
1481 lines (1323 loc) Β· 60.9 KB
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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# AmpliPi Home Audio
# Copyright (C) 2022 MicroNova LLC
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Control of the whole AmpliPi Audio System.
This module provides complete control of the AmpliPi Audio System's sources,
zones, groups and streams.
"""
from typing import List, Dict, Set, Union, Optional, Callable
from enum import Enum
from copy import deepcopy
import os # files
from pathlib import Path
import time
import logging
import sys
import datetime
import psutil
import threading
import wrapt
from amplipi import models
from amplipi import rt
from amplipi import utils
import amplipi.streams
from amplipi.eeprom import EEPROM, BoardType, find_boards
from amplipi import auth
from amplipi import defaults
import traceback
from amplipi.streams.base_streams import PersistentStream, InvalidStreamField
_DEBUG_API = False # print out a graphical state of the api after each call
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
sh = logging.StreamHandler(sys.stdout)
logger.addHandler(sh)
@wrapt.decorator
def save_on_success(wrapped, instance: 'Api', args, kwargs):
""" Check if a ctrl API call is successful and saves the state if so """
result = wrapped(*args, **kwargs)
if result is None:
# call mark_changes instead of save to reduce the load/delay of a series of requests
instance.mark_changes()
return result
class ApiCode(Enum):
""" Ctrl Api Response code """
OK = 1
ERROR = 2
class ApiResponse:
""" Ctrl Api Response object """
def __init__(self, code: ApiCode, msg: str = '', data: Optional[Dict] = None):
self.code = code
self.msg = msg
if data:
self.data = data
else:
self.data = {}
def __str__(self):
if self.code == ApiCode.OK:
return 'OK'
return f'ERROR: {self.msg}'
@staticmethod
def error(msg: str):
""" create an error response """
return ApiResponse(ApiCode.ERROR, msg)
@staticmethod
def fieldError(field: str, msg: str):
""" create an error response for a specific field """
return ApiResponse(ApiCode.ERROR, msg, {"field": field})
@staticmethod
def ok():
""" create a successful response """
return ApiResponse(ApiCode.OK)
OK = ApiCode.OK
ERROR = ApiCode.ERROR
class Api:
""" Amplipi Controller API"""
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-many-public-methods
# TODO: move these variables to the init, they should not be class variables.
_initialized = False # we need to know when we initialized first
_mock_hw: bool
_mock_streams: bool
_save_timer: Optional[threading.Timer] = None
_delay_saves: bool
_change_notifier: Optional[Callable[[models.Status], None]] = None
_rt: Union[rt.Rpi, rt.Mock]
config_file: str
backup_config_file: str
config_file_valid: bool
is_streamer: bool
status: models.Status
streams: Dict[int, amplipi.streams.AnyStream]
lms_mode: bool
_serial: Optional[int] = None
_expanders: List[int] = []
_freeze_delete_temporary: bool = False
# TODO: migrate to init setting instance vars to a disconnected state (API requests will throw Api.DisconnectedException() in this state
# with this reinit will be called connect and will attempt to load the configuration and connect to an AmpliPi (mocked or real)
# returning a boolean on whether or not it was successful
def __init__(self, settings: models.AppSettings = models.AppSettings(), change_notifier: Optional[Callable[[models.Status], None]] = None):
self.reinit(settings, change_notifier)
self._initialized = True
def reinit(self, settings: models.AppSettings = models.AppSettings(), change_notifier: Optional[Callable[[models.Status], None]] = None, config: Optional[models.Status] = None):
""" Initialize or Reinitialize the controller
Initializes the system to to base configuration """
self._change_notifier = change_notifier
self._mock_hw = settings.mock_ctrl
self._mock_streams = settings.mock_streams
self._save_timer = None
self._delay_saves = settings.delay_saves
self._settings = settings
# try to get a list of available boards to determine if we are a streamer
# the preamp hardware is not available on a streamer
# we need to know this before trying to initialize the firmware
found_boards = []
try:
found_boards = find_boards()
if found_boards:
logger.info(f'Found boards:')
except Exception as exc:
logger.exception(f'Error finding boards: {exc}')
try:
for board in found_boards:
board_info = EEPROM(board).read_board_info()
logger.info(f' - {board_info}')
if self._serial is None and board_info is not None:
self._serial = board_info.serial
except Exception as exc:
logger.exception(f'Error showing board info: {exc}')
# check if we are a streamer
self.is_streamer = BoardType.STREAMER_SUPPORT in found_boards
# Create firmware interface if needed. If one already exists delete then re-init.
if self._initialized:
# we need to make sure to mute every zone before resetting the fw
zones_update = models.MultiZoneUpdate(zones=[z.id for z in self.status.zones],
update=models.ZoneUpdate(mute=True))
try:
self.set_zones(zones_update, force_update=True, internal=True)
except OSError:
logger.info("Failed to mute some of the zones before resetting.")
try:
del self._rt # remove the low level hardware connection
except AttributeError:
pass
self._rt = rt.Mock() if settings.mock_ctrl or self.is_streamer else rt.Rpi() # reset the fw
# test open the config file, this will throw an exception if there are issues writing to the file
# use append more to make sure we have read and write permissions, but won't overwrite the file
with open(settings.config_file, 'a', encoding='utf-8'):
pass
self.config_file = settings.config_file
self.backup_config_file = settings.config_file + '.bak'
self.config_file_valid = True # initially we assume the config file is valid
errors = []
if config:
self.status = config
loaded_config = True
else:
# try to load the config file or its backup
config_paths = [self.config_file, self.backup_config_file]
loaded_config = False
for cfg_path in config_paths:
try:
if os.path.exists(cfg_path):
self.status = models.Status.parse_file(cfg_path)
loaded_config = True
break
errors.append(f'config file "{cfg_path}" does not exist')
except Exception as exc:
self.config_file_valid = False # mark the config file as invalid so we don't try to back it up
errors.append(f'error loading config file: {exc}')
# make a config flag to recognize this unit's subtype
# this helps the updater make good decisions
is_streamer_path = Path(defaults.USER_CONFIG_DIR, 'is_streamer')
try:
if self.is_streamer:
is_streamer_path.touch()
elif is_streamer_path.exists():
os.remove(is_streamer_path)
except Exception as exc:
logger.exception("Error setting is_streamer flag: {exc}")
# determine if we're in LMS mode, based on a file existing
lms_mode_path = Path(defaults.USER_CONFIG_DIR, 'lms_mode')
if lms_mode_path.exists():
logger.info("lms mode")
self.lms_mode = True
else:
logger.info("not lms mode")
self.lms_mode = False
# load a good default config depending on the unit subtype
if not loaded_config:
if len(errors) > 0:
logger.error(errors[0])
default_config = defaults.default_config(is_streamer=self.is_streamer, lms_mode=self.lms_mode)
self.status = models.Status.parse_obj(default_config)
self.save()
# populate system info
self._online_cache = utils.TimeBasedCache(self._check_is_online, 5, 'online')
self._latest_release_cache = utils.TimeBasedCache(self._check_latest_release, 3600, 'latest release')
self._connected_drives_cache = utils.TimeBasedCache(self._get_usb_drives, 5, 'connected drives')
self.status.info = models.Info(
mock_ctrl=self._mock_hw,
mock_streams=self._mock_streams,
config_file=self.config_file,
is_streamer=self.is_streamer,
lms_mode=self.lms_mode,
version=utils.detect_version(),
stream_types_available=amplipi.streams.stream_types_available(),
extra_fields=utils.load_extra_fields(),
serial=str(self._serial)
)
for major, minor, ghash, dirty in self._rt.read_versions():
fw_info = models.FirmwareInfo(version=f'{major}.{minor}', git_hash=f'{ghash:x}', git_dirty=dirty)
self.status.info.fw.append(fw_info)
if self.status.info.fw:
fw_info = self.status.info.fw[0]
logger.info(f"Main unit firmware version: {fw_info.version}")
else:
logger.info(f"No preamp")
if self.status.info.fw[1:]:
for i, fw_info in enumerate(self.status.info.fw[1:], start=1):
logger.info(f"Expansion unit {i} firmware version: {fw_info.version}")
else:
logger.info(f"No expansion units")
self._update_sys_info() # TODO: does sys info need to be updated at init time?
# detect missing zones
if self._mock_hw and not self.is_streamer:
# only allow 6 zones when mocked to simplify testing
# add more if needed by specifying them in the config
potential_zones = range(6)
elif self.is_streamer:
# streamer has no zones
potential_zones = range(0)
else:
potential_zones = range(rt.MAX_ZONES)
added_zone = False
for zid in potential_zones:
_, zone = utils.find(self.status.zones, zid)
if zone is None and self._rt.exists(zid):
added_zone = True
self.status.zones.append(models.Zone(id=zid, name=f'Zone {zid+1}'))
# save new config if zones were added
if added_zone:
self.save()
# populate mute_all preset with all zones
_, mute_all_pst = utils.find(self.status.presets, defaults.MUTE_ALL_ID)
if mute_all_pst and mute_all_pst.name == 'Mute All':
if mute_all_pst.state and mute_all_pst.state.zones:
muted_zones = {z.id for z in mute_all_pst.state.zones}
for z in self.status.zones:
if z.id not in muted_zones:
mute_all_pst.state.zones.append(models.ZoneUpdateWithId(id=z.id, mute=True))
# configure Aux and SPDIF
utils.configure_inputs()
if not self.is_streamer:
# add any missing RCA stream, mostly used to migrate old configs where rca inputs were not streams
for rca_id in defaults.RCAs:
sid, stream = utils.find(self.status.streams, rca_id)
if sid is None:
idx = rca_id - defaults.RCAs[0]
# try to use the old name in the source if it was renamed from the default name of 1-4
input_name = f'Input {idx + 1}'
try:
src_name = self.status.sources[idx].name
if not src_name.isdigit():
input_name = src_name
except Exception as e:
logger.exception(f'Error discovering old source name for conversion to RCA stream: {e}')
logger.info(f'- Defaulting name to: {input_name}')
rca_stream = models.Stream(id=rca_id, name=input_name, type='rca', index=idx)
self.status.streams.insert(idx, rca_stream)
# make sure the config file contains the aux stream
has_aux_stream = False
for stream in self.status.streams:
if stream.type == "aux":
has_aux_stream = True
break
if not has_aux_stream:
# insert aux stream in appropriate place
self.status.streams.insert(0, models.Stream(id=defaults.AUX_STREAM_ID, type="aux", name="Aux"))
# configure all streams into a known state
self.streams: Dict[int, amplipi.streams.AnyStream] = {}
failed_streams: List[int] = []
for stream in self.status.streams:
assert stream.id is not None
if stream.id:
try:
self.streams[stream.id] = amplipi.streams.build_stream(stream, self._mock_streams, validate=False)
# If we're in LMS mode, we need to start these clients on each boot, not when they get assigned to a
# particular source; the client+server connection bootstrapping takes a while, which is a less than ideal
# user experience.
if self.lms_mode and stream.type == 'lms':
self.streams[stream.id].activate() # type: ignore
except Exception as exc:
logger.exception(f"Failed to create '{stream.name}' stream: {exc}")
failed_streams.append(stream.id)
self.sync_stream_info() # need to update the status with the new streams
# add/remove dynamic bluetooth stream
bt_streams = [sid for sid, stream in self.streams.items() if isinstance(stream, amplipi.streams.Bluetooth)]
if amplipi.streams.Bluetooth.is_hw_available() and not self._mock_hw:
logger.info('bluetooth dongle available')
# make sure one stream is available
if len(bt_streams) == 0:
logger.info('no bt streams present. creating one')
self.create_stream(models.Stream(type='bluetooth', name='Bluetooth'), internal=True)
elif len(bt_streams) > 1:
logger.info('bt streams present. removing all but one')
for s in bt_streams[1:]:
self.delete_stream(s, internal=True)
else:
logger.info('bluetooth dongle unavailable')
if len(bt_streams) > 0:
logger.info('bt streams present. removing all')
for s in bt_streams:
self.delete_stream(s, internal=True)
# enable/disable any FMRadio streams, depending on hw availability
fm_streams = [stream for sid, stream in self.streams.items() if isinstance(stream, amplipi.streams.FMRadio)]
fm_disabled = not amplipi.streams.FMRadio.is_hw_available()
if fm_disabled:
logger.warning('fm radio dongle unavailable')
for fm_stream in fm_streams:
logger.info(f"setting FM stream {fm_stream.name} to disabled={fm_disabled} based on hw availability")
fm_stream.disabled = fm_disabled
self.sync_stream_info() # update stream status with potentially updated streams
# configure all sources so that they are in a known state
# only models.MAX_SOURCES are supported, keep the config from adding extra
# this helps us transition from weird and experimental configs
try:
self.status.sources[:] = self.status.sources[0:models.MAX_SOURCES]
except Exception as exc:
logger.exception('Error configuring sources: using all defaults')
self.status.sources[:] = [models.Source(id=i, name=f'Input {i+1}') for i in range(models.MAX_SOURCES)]
# populate any missing sources, to match the underlying system's capabilities
for sid in range(len(self.status.sources), models.MAX_SOURCES):
logger.warning(f'Error: missing source {sid}, inserting default source')
self.status.sources.insert(sid, models.Source(id=sid, name=f'Input {sid+1}'))
# sequentially number sources if necessary
for sid, src in enumerate(self.status.sources):
if src.id != sid:
logger.info(f'Error: source at index {sid} is not sequentially numbered, fixing')
src.id = sid
# configure all of the sources, now that they are layed out as expected
for src in self.status.sources:
if src.id is not None:
try:
update = models.SourceUpdate(input=src.input)
self.set_source(src.id, update, force_update=True, internal=True)
except Exception as e:
logger.exception(f'Error configuring source {src.id}: {e}')
logger.info(f'defaulting source {src.id} to an empty input')
update = models.SourceUpdate(input='')
try:
self.set_source(src.id, update, force_update=True, internal=True)
except Exception as e_empty:
logger.exception(f'Error configuring source {src.id}: {e_empty}')
logger.info(f'Source {src.id} left uninitialized')
# configure all of the zones so that they are in a known state
# we mute all zones on startup to keep audio from playing immediately at startup
for zone in self.status.zones:
# TODO: disable zones that are not found
# we likely need an additional field for this, maybe auto-disabled?
zone_update = models.ZoneUpdate(source_id=zone.source_id, mute=True, vol=zone.vol)
self.set_zone(zone.id, zone_update, force_update=True, internal=True)
# configure all of the groups (some fields may need to be updated)
self._update_groups()
def __del__(self):
# stop save in the future so we can save right away
# we save before shutting everything down to avoid saving disconnected state
if self._save_timer:
self._save_timer.cancel()
self._save_timer = None
self.save()
# stop any streams
for stream in self.streams.values():
stream.disconnect()
# lower the volume on any unmuted zone to limit popping
# behind the scenes the firmware will gradually lower the volume until the output is effectively muted
# muting is more likely to cause popping
# we use the low level rt calls to avoid changing the configuration
vol_changes = False
for z in self.status.zones:
if not z.mute:
self._rt.update_zone_vol(z.id, models.MIN_VOL_DB)
vol_changes = True
if vol_changes:
# wait for the changes to take effect (we observed a tiny pop without this)
time.sleep(0.080)
# put the firmware in a reset state
self._rt.reset()
def save(self) -> None:
""" Saves the system state to json"""
try:
# save a backup copy of the config file (assuming its valid)
if os.path.exists(self.config_file) and self.config_file_valid:
if os.path.exists(self.backup_config_file):
os.remove(self.backup_config_file)
os.rename(self.config_file, self.backup_config_file)
with open(self.config_file, 'w', encoding='utf-8') as cfg:
cfg.write(self.status.json(exclude_none=True, indent=2))
self.config_file_valid = True
except Exception as exc:
logger.exception(f'Error saving config: {exc}')
def mark_changes(self):
""" Mark api changes to update listeners and save the system state in the future
This attempts to avoid excessive saving and the resulting delays by only saving a small delay after the last change
"""
if self._change_notifier:
self._change_notifier(self.get_state())
if self._delay_saves:
if self._save_timer:
self._save_timer.cancel()
self._save_timer = None
# start can only be called once on a thread
self._save_timer = threading.Timer(5.0, self.save)
self._save_timer.start()
else:
self.save()
def _is_digital(self, sinput: str) -> bool:
"""Determines whether a source input, @sinput, is analog or digital
@sinput is expected to be one of the following:
| str | meaning | analog or digital? |
| -------------------- | -------- | ------------------ |
| '' | no input | digital |
| 'stream={stream_id}' | a stream | analog or digital (depending on stream_id's stream type) |
The runtime only has the concept of digital or analog.
The system defaults to digital as an undriven analog input acts as an antenna,
producing a small amount of white noise.
"""
try:
sid = int(sinput.replace('stream=', ''))
return sid not in defaults.RCAs
except:
return True
def get_inputs(self, src: models.Source) -> Dict[Optional[str], str]:
"""Gets a dictionary of the possible inputs for a source
Returns:
A dictionary of the input types and a corresponding user friendly name/string for each
Example:
Get the possible inputs for any source (only one stream)
>>> my_amplipi.get_inputs()
{ '': '', 'stream=9449': 'Matt and Kim Radio' }
"""
inputs: Dict[Optional[str], str] = {'': ''}
for sid, stream in self.streams.items():
# TODO: remove this filter when sources can dynamically change output
connectable = stream.requires_src() in [None, src.id]
connected = src.get_stream()
if sid == connected:
assert connectable, print(f'Source {src} has invalid input: stream={connected}')
if (sid == connected or not stream.disabled) and connectable:
inputs[f'stream={sid}'] = stream.full_name()
return inputs
def _check_is_online(self) -> bool:
online = False
try:
with open(os.path.join(defaults.USER_CONFIG_DIR, 'online'), encoding='utf-8') as fonline:
online = 'online' in fonline.readline()
except Exception:
pass
return online
def _check_latest_release(self) -> str:
release = 'unknown'
try:
with open(os.path.join(defaults.USER_CONFIG_DIR, 'latest_release'), encoding='utf-8') as flatest:
release = flatest.readline().strip()
except Exception:
pass
return release
def _get_usb_drives(self):
"""Reads the mount point for removable drives, returns them in a list"""
usb_drives = []
for partition in psutil.disk_partitions():
# Exclude rootfs, the backup drive that happens to mount to /media/pi
if '/media' in partition.mountpoint and 'rootfs' not in partition.mountpoint:
usb_drives.append(str(partition.mountpoint))
return usb_drives
def _update_sys_info(self, throttled=True) -> None:
"""Update current system information"""
if self.status.info is None:
raise Exception("No info generated, system in a bad state")
self.status.info.online = self._online_cache.get(throttled)
self.status.info.connected_drives = self._connected_drives_cache.get(throttled)
self.status.info.latest_release = self._latest_release_cache.get(throttled)
self.status.info.access_key = auth.get_access_key("admin") if auth.user_access_key_set("admin") else ""
def sync_stream_info(self) -> None:
"""Synchronize the stream list to the stream status"""
# TODO: figure out how to cache stream info, since it only needs to happen when a stream is added/updated
streams = []
for sid, stream_inst in self.streams.items():
# TODO: this functionality should be in the unimplemented streams base class
# convert the stream instance info to stream data (serialize its current configuration)
st_type = type(stream_inst).stream_type
stream = models.Stream(id=sid, name=stream_inst.name, type=st_type)
for field in models.optional_stream_fields():
if field in stream_inst.__dict__:
stream.__dict__[field] = stream_inst.__dict__[field]
streams.append(stream)
self.status.streams = streams
def _update_serial(self) -> None:
expanders = []
# Preamp Boards
eeprom_preamp = EEPROM(BoardType.PREAMP)
eeprom_streamer = EEPROM(BoardType.STREAMER_SUPPORT)
if eeprom_preamp.present():
bi = eeprom_preamp.read_board_info()
if bi is not None:
self._serial = int(bi.serial)
for board in rt.get_dev_addrs()[1:]: # Check for expanders
if eeprom_preamp.present(board):
exp_info = eeprom_preamp.read_board_info(board)
if exp_info is not None:
expanders.append(exp_info.serial)
self._expanders = expanders
if self.status.info is not None:
self.status.info.expanders = expanders
elif eeprom_streamer.present(): # Streamers
bi = eeprom_streamer.read_board_info()
if bi is not None:
self._serial = bi.serial
else: # -1 means there is no serial number available from any EEPROM
self._serial = -1
if self.status.info is not None:
self.status.info.serial = str(self._serial)
def get_state(self) -> models.Status:
""" get the system state """
self._update_sys_info()
if not self._freeze_delete_temporary:
self._delete_unused_temporary_streams()
# Get serial number
if self._serial is None and self.status.info is not None:
self._update_serial()
# update source's info
# TODO: source info should be updated in a background thread
for src in self.status.sources:
self._update_src_info(src)
return self.status
def get_info(self) -> models.Info:
""" Get the system information """
self._update_sys_info()
if self.status.info is None:
raise Exception("No info generated, system in a bad state")
return self.status.info
def get_items(self, tag: str) -> Optional[List[models.Base]]:
""" Gets one of the lists of elements contained in status named by @t (or t's plural
This makes it easy to programmatically access zones using t='zones' or groups using t='groups'.
We use this to generate some documentation.
"""
item_lists = ['streams', 'presets', 'sources', 'zones', 'groups']
plural_tag = tag + 's'
items = []
if tag in item_lists:
items = self.get_state().__dict__[tag]
elif plural_tag in item_lists:
items = self.get_state().__dict__[plural_tag]
return items
def get_stream(self, src: Optional[models.Source] = None, sid: Optional[int] = None) -> Optional[amplipi.streams.AnyStream]:
"""Gets the stream from a source
Args:
src: An audio source that may have a stream connected
sid: ID of an audio source
Returns:
a stream, or None if input does not specify a valid stream
"""
if sid is not None:
_, src = utils.find(self.status.sources, sid)
if src is None:
return None
idx = src.get_stream()
if idx is not None:
return self.streams.get(idx, None)
return None
def _update_src_info(self, src: models.Source):
""" Update a source's status and song metadata """
stream_inst = self.get_stream(src)
if stream_inst is not None:
src.info = stream_inst.info()
else:
src.info = models.SourceInfo(img_url='static/imgs/disconnected.png', name='None', state='stopped')
def _get_source_config(self, sources: Optional[List[models.Source]] = None) -> List[bool]:
""" Convert the preamp's source configuration """
if not sources:
sources = self.status.sources
src_cfg = [True] * models.MAX_SOURCES
for s, src in enumerate(sources):
src_cfg[s] = self._is_digital(src.input)
return src_cfg
def _delete_unused_temporary_streams(self):
"""Removes temporary file players if they are disconnected and have no connected sources"""
temp_streams = []
for stream_id in self.streams.keys():
stream = self.streams[stream_id]
if stream.stream_type == 'fileplayer' and stream.temporary and stream.timeout_expired():
temp_streams.append(stream_id)
for source in self.status.sources:
for stream_id in temp_streams:
if source.input[7:].isdigit() and int(source.input[7:]) == stream_id:
temp_streams.remove(stream_id)
for stream_id in temp_streams:
logger.info(f'Deleting unused temporary stream {stream_id}')
self.delete_stream(stream_id, internal=False) # Internal is False so it shows up immediately on UI
def set_source(self, sid: int, update: models.SourceUpdate, force_update: bool = False, internal: bool = False) -> ApiResponse:
"""Modifes the configuration of one of the 4 system sources
Args:
sid (int): source id [0,3]
update: changes to source
force_update: bool, update source even if no changes have been made (for hw startup)
internal: called by a higher-level ctrl function:
Returns:
'None' on success, otherwise error (dict)
"""
try:
idx, src = utils.find(self.status.sources, sid)
if idx is not None and src is not None:
name, _ = utils.updated_val(update.name, src.name)
input_, input_updated = utils.updated_val(update.input, src.input)
# update the name
src.name = str(name)
if input_updated or force_update:
# shutdown old stream
old_stream = self.get_stream(src)
if old_stream and old_stream.is_connected():
old_stream.disconnect()
# start new stream
last_input = src.input
src.input = input_ # reconfigure the input so get_stream knows which stream to get
stream = self.get_stream(src)
if stream:
if stream.disabled:
raise Exception(f"Stream {stream.name} is disabled")
stolen_from: Optional[models.Source] = None
if stream.src is not None and stream.src != idx:
# update the streams last connected source to have no input, since we have stolen its input
stolen_from = self.status.sources[stream.src]
logger.info(f'stealing {stream.name} from source {stolen_from.name}')
stolen_from.input = ''
try:
if stream.is_connected():
stream.disconnect()
stream.connect(idx)
# potentially deactivate the old stream to save resources
# NOTE: old_stream and new stream could be the same if force_update is True
if old_stream and old_stream != stream and old_stream.is_activated():
if not old_stream.is_persistent(): # type: ignore
old_stream.deactivate() # type: ignore
except Exception as iexc:
logger.exception(f"Failed to update {sid}'s input to {stream.name}: {iexc}")
stream.disconnect()
if old_stream:
logger.info(f'Trying to get back to the previous input: {old_stream.name}')
old_stream.connect(idx)
src.input = last_input
else:
src.input = ''
# connect the stream back to its old source
if stolen_from and stolen_from.id is not None:
logger.info(f"Trying to revert src {stolen_from.id}'s input to {stream.name}")
stream.connect(stolen_from.id)
stolen_from.input = input_
# now that we recovered, show that this failed
raise iexc
elif src.input and 'stream=' in src.input: # invalid stream id?
# TODO: should this stream id validation happen in the Source model?
src.input = last_input
raise Exception(f'StreamID specified by "{src.input}" not found')
elif old_stream and old_stream.is_activated():
if not old_stream.is_persistent(): # type: ignore
old_stream.deactivate() # type: ignore
if not self.is_streamer:
rt_needs_update = self._is_digital(input_) != self._is_digital(last_input)
if rt_needs_update or force_update:
src_cfg = self._get_source_config()
if not self._rt.update_sources(src_cfg):
raise Exception('failed to set source')
self._update_src_info(src) # synchronize the source's info
if not internal:
self.mark_changes()
else:
raise Exception(f'failed to set source: index {idx} out of bounds')
except Exception as exc:
if internal:
raise exc
return ApiResponse.error(f'failed to set source: {exc}')
return ApiResponse.ok()
def set_zone(self, zid, update: models.ZoneUpdate, force_update: bool = False, internal: bool = False) -> ApiResponse:
"""Reconfigures a zone
Args:
id: any valid zone [0,p*6-1] (6 zones per preamp)
update: changes to zone
force_update: update source even if no changes have been made (for hw startup)
internal: called by a higher-level ctrl function
Returns:
ApiResponse
"""
try:
idx, zone = utils.find(self.status.zones, zid)
if idx is not None and zone is not None:
name, _ = utils.updated_val(update.name, zone.name)
source_id, update_source_id = utils.updated_val(update.source_id, zone.source_id)
mute, update_mutes = utils.updated_val(update.mute, zone.mute)
vol, update_vol = utils.updated_val(update.vol, zone.vol)
vol_f, update_vol_f = utils.updated_val(update.vol_f, zone.vol_f)
vol_delta_f = update.vol_delta_f
vol_min, update_vol_min = utils.updated_val(update.vol_min, zone.vol_min)
vol_max, update_vol_max = utils.updated_val(update.vol_max, zone.vol_max)
disabled, _ = utils.updated_val(update.disabled, zone.disabled)
# update non hw state
zone.name = name
zone.disabled = disabled
# parse and check the source id
sid = utils.parse_int(source_id, range(models.ZONE_OFF, models.MAX_SOURCES))
special_status_sid = sid in [models.ZONE_OFF, models.SOURCE_DISCONNECTED]
# disconnect zone from source if it's disabled.
if zone.disabled:
sid = models.SOURCE_DISCONNECTED
update_source_id = True
# any zone disabled by source disconnection or a 'disabled' flag should not be able to play anything
implicit_mute = zone.disabled or special_status_sid
if implicit_mute and not mute:
mute = True
update_mutes = True
# update the zone's associated source
zones = self.status.zones
if update_source_id or force_update:
# the preamp fw needs nearby zones source-ids since each source id register contains the source ids of 3 zones
zone_sources = [utils.clamp(zone.source_id, 0, 3) for zone in zones]
# update with the pending change
zone_sources[zid] = utils.clamp(source_id, 0, 3)
# this is setting the state for all zones
# TODO: cache the fw state and only do this on change, this quickly gets out of hand when changing many zones
if special_status_sid:
# don't send the source id to the firmware if we are disconnecting the source
zone.source_id = sid
elif self._rt.update_zone_sources(idx, zone_sources):
zone.source_id = sid
else:
raise Exception('unable to update zone source')
# update min/max volumes
if vol_max - vol_min < models.MIN_DB_RANGE:
raise Exception(f'max - min volume must be greater than {models.MIN_DB_RANGE}')
zone.vol_min = vol_min
zone.vol_max = vol_max
def set_mute():
mutes = [zone.mute for zone in zones]
mutes[idx] = mute
if self._rt.update_zone_mutes(idx, mutes):
zone.mute = mute
else:
raise Exception('unable to update zone mute')
def set_vol():
""" Update the zone's volume. Could be triggered by a change in
vol, vol_f, vol_f_delta, vol_min, or vol_max.
"""
# Field precedence: vol (db) > vol_delta > vol (float)
# vol (db) is first in precedence yet last in the stack to cover the default case of no volume change
if update.vol_delta_f is not None and update.vol is None:
true_vol_f = zone.vol_f + zone.vol_f_overflow
expected_vol_total = update.vol_delta_f + true_vol_f
vol_f_new = utils.clamp(expected_vol_total, models.MIN_VOL_F, models.MAX_VOL_F)
vol_db = utils.vol_float_to_db(vol_f_new, zone.vol_min, zone.vol_max)
zone.vol_f_overflow = 0 if models.MIN_VOL_F < expected_vol_total and expected_vol_total < models.MAX_VOL_F \
else utils.clamp((expected_vol_total - vol_f_new), models.MIN_VOL_F_OVERFLOW, models.MAX_VOL_F_OVERFLOW)
# Clamp the remaining delta to be between -1 and 1
elif update.vol_f is not None and update.vol is None:
clamp_vol_f = utils.clamp(vol_f, 0, 1)
vol_db = utils.vol_float_to_db(clamp_vol_f, zone.vol_min, zone.vol_max)
vol_f_new = clamp_vol_f
else:
vol_db = utils.clamp(vol, zone.vol_min, zone.vol_max)
vol_f_new = utils.vol_db_to_float(vol_db, zone.vol_min, zone.vol_max)
if self._rt.update_zone_vol(idx, vol_db):
zone.vol = vol_db
zone.vol_f = vol_f_new
else:
raise Exception('unable to update zone volume')
# Reset the overflow when vol_f goes in bounds, there is no longer an overflow
# Avoids reporting spurious volume oscillations
zone.vol_f_overflow = 0 if vol_f_new != models.MIN_VOL_F and vol_f_new != models.MAX_VOL_F else zone.vol_f_overflow
# To avoid potential unwanted loud output:
# If muting, mute before setting volumes
# If un-muting, set desired volume first
update_volumes = update_vol or update_vol_f or update_vol_min or update_vol_max or vol_delta_f is not None
if force_update or (update_mutes and update_volumes):
if mute:
set_mute()
set_vol()
else:
set_vol()
set_mute()
elif update_volumes:
set_vol()
elif update_mutes:
set_mute()
if not internal:
# update the group stats (individual zone volumes, sources, and mute configuration can effect a group)
self._update_groups()
self.mark_changes()
except Exception as exc:
if internal:
raise exc
return ApiResponse.error(f'set zone failed: {exc}')
else:
return ApiResponse.ok()
def set_zones(self, multi_update: models.MultiZoneUpdate, force_update: bool = False, internal: bool = False) -> ApiResponse:
"""Reconfigures a set of zones
Args:
update: changes to apply to embedded zones and groups
Returns:
ApiResponse
"""
try:
# aggregate all of the zones together
all_zids = utils.zones_from_all(self.status, multi_update.zones, multi_update.groups)
# update each of the zones
for zid in all_zids:
zupdate = multi_update.update.copy() # we potentially need to make changes to the underlying update
if zupdate.name:
# ensure all zones don't get named the same
zupdate.name = f'{zupdate.name} {zid+1}'
self.set_zone(zid, zupdate, force_update=force_update, internal=True)
if not internal:
# update the group stats (individual zone volumes, sources, and mute configuration can effect a group)
self._update_groups()
self.mark_changes()
except Exception as exc:
if internal:
raise exc
return ApiResponse.error(f'set_zones failed: {exc}')
return ApiResponse.ok()
def _update_groups(self) -> None:
"""Updates the group's aggregate fields to maintain consistency and simplify app interface"""
for group in self.status.groups:
zones = [self.status.zones[z] for z in group.zones]
# remove disabled from further calculations
zones = [z for z in zones if not z.disabled]
mutes = [z.mute for z in zones]
sources = {z.source_id for z in zones}
vols = [z.vol_f for z in zones]
vols.sort()
group.mute = False not in mutes # group is only considered muted if all zones are muted
if len(sources) == 1:
group.source_id = sources.pop() # TODO: how should we handle different sources in the group?
else: # multiple sources
group.source_id = None
if vols:
group.vol_f = (vols[0] + vols[-1]) / 2 # group volume is the midpoint between the highest and lowest source
else:
group.vol_f = models.MIN_VOL_F
group.vol_delta = utils.vol_float_to_db(group.vol_f)
def set_group(self, gid, update: models.GroupUpdate, internal: bool = False) -> ApiResponse:
"""Configures an existing group
parameters will be used to configure each zone in the group's zones
all parameters besides the group id, @id, are optional
Args:
gid: group id (a guid)
update: changes to group
internal: called by a higher-level ctrl function
Returns:
'None' on success, otherwise error (dict)
"""
try:
_, group = utils.find(self.status.groups, gid)
if group is None:
raise Exception(f'group {gid} not found')
name, _ = utils.updated_val(update.name, group.name)
zones, _ = utils.updated_val(update.zones, group.zones)
vol_delta, vol_updated = utils.updated_val(update.vol_delta, group.vol_delta)
vol_f, vol_f_updated = utils.updated_val(update.vol_f, group.vol_f)
group.name = name
group.zones = zones
# determine group volume, 'vol_delta' (in dB) takes precedence over vol_f
# if vol_updated is true vol_delta can't be none but mypy isn't smart enough to know that
if vol_updated and vol_delta is not None:
vol_f = utils.vol_db_to_float(vol_delta)
# update each of the member zones
zone_update = models.ZoneUpdate(source_id=update.source_id, mute=update.mute)
if vol_updated or vol_f_updated:
# TODO: make this use volume delta adjustment, for now its a fixed group volume
# use float value so zone calculates appropriate offsets in dB
zone_update.vol_f = vol_f
for zone in [self.status.zones[zone] for zone in zones]:
self.set_zone(zone.id, zone_update, internal=True)
if not internal:
# update the group stats