/
denonavr.py
1752 lines (1542 loc) · 69.1 KB
/
denonavr.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
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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
This module implements the interface to Denon AVR receivers.
:copyright: (c) 2016 by Oliver Goetz.
:license: MIT, see LICENSE for more details.
"""
from collections import (namedtuple, OrderedDict)
from io import BytesIO
import logging
import time
import re
import html
import xml.etree.ElementTree as ET
import requests
_LOGGER = logging.getLogger("DenonAVR")
DEVICEINFO_AVR_X_PATTERN = re.compile(
r"(.*AVR-(X|S).*|.*SR500[6-9]|.*SR60(07|08|09|10|11|12|13)|.*NR1604)")
DEVICEINFO_COMMAPI_PATTERN = re.compile(r"(0210|0300)")
ReceiverType = namedtuple('ReceiverType', ["type", "port"])
AVR = ReceiverType(type="avr", port=80)
AVR_X = ReceiverType(type="avr-x", port=80)
AVR_X_2016 = ReceiverType(type="avr-x-2016", port=8080)
SOURCE_MAPPING = {"TV AUDIO": "TV", "iPod/USB": "USB/IPOD", "Bluetooth": "BT",
"Blu-ray": "BD", "CBL/SAT": "SAT/CBL", "NETWORK": "NET",
"Media Player": "MPLAY", "AUX": "AUX1", "Tuner": "TUNER",
"FM": "TUNER"}
CHANGE_INPUT_MAPPING = {"Internet Radio": "IRP", "Online Music": "NET",
"Media Server": "SERVER", "Spotify": "SPOTIFY",
"Flickr": "FLICKR", "Favorites": "FAVORITES"}
ALL_ZONE_STEREO = "ALL ZONE STEREO"
SOUND_MODE_MAPPING = OrderedDict(
[('MUSIC', ['PLII MUSIC', 'DTS NEO:6 MUSIC', 'DOLBY D +NEO:X M']),
('MOVIE', ['PLII MOVIE', 'PLII CINEMA', 'DTS NEO:X CINEMA',
'DTS NEO:6 CINEMA', 'DOLBY D +NEO:X C',
'PLIIX CINEMA']),
('GAME', ['PLII GAME', 'DOLBY D +NEO:X G']),
('AUTO', ['None']),
('STANDARD', ['None2']),
('VIRTUAL', ['VIRTUAL']),
('MATRIX', ['MATRIX']),
('ROCK ARENA', ['ROCK ARENA']),
('JAZZ CLUB', ['JAZZ CLUB']),
('VIDEO GAME', ['VIDEO GAME']),
('MONO MOVIE', ['MONO MOVIE']),
('DIRECT', ['DIRECT']),
('PURE DIRECT', ['PURE_DIRECT', 'PURE DIRECT']),
('DOLBY DIGITAL', ['DOLBY DIGITAL', 'DOLBY D + DOLBY SURROUND',
'DOLBY DIGITAL +', 'STANDARD(DOLBY)', 'DOLBY SURROUND',
'DOLBY D + +DOLBY SURROUND', 'NEURAL', 'DOLBY HD',
'MULTI IN + NEURAL:X', 'MULTI IN + DOLBY SURROUND',
'DOLBY D + NEURAL:X', 'DOLBY DIGITAL + NEURAL:X',
'DOLBY DIGITAL + + NEURAL:X', 'DOLBY ATMOS',
'DOLBY AUDIO - DOLBY SURROUND', 'DOLBY TRUEHD']),
('DTS SURROUND', ['DTS SURROUND', 'DTS NEURAL:X', 'STANDARD(DTS)',
'DTS + NEURAL:X', 'MULTI CH IN', 'DTS-HD MSTR',
'DTS VIRTUAL:X']),
('MCH STEREO', ['MULTI CH STEREO', 'MULTI_CH_STEREO', 'MCH STEREO']),
('STEREO', ['STEREO']),
(ALL_ZONE_STEREO, [ALL_ZONE_STEREO])])
PLAYING_SOURCES = ("Online Music", "Media Server", "iPod/USB", "Bluetooth",
"Internet Radio", "Favorites", "SpotifyConnect", "Flickr",
"TUNER", "NET/USB", "HDRADIO", "Music Server", "NETWORK",
"NET")
NETAUDIO_SOURCES = ("Online Music", "Media Server", "iPod/USB", "Bluetooth",
"Internet Radio", "Favorites", "SpotifyConnect", "Flickr",
"NET/USB", "Music Server", "NETWORK", "NET")
# Image URLs
STATIC_ALBUM_URL = "http://{host}:{port}/img/album%20art_S.png"
ALBUM_COVERS_URL = "http://{host}:{port}/NetAudio/art.asp-jpg?{time}"
# General URLs
APPCOMMAND_URL = "/goform/AppCommand.xml"
DEVICEINFO_URL = "/goform/Deviceinfo.xml"
NETAUDIOSTATUS_URL = "/goform/formNetAudio_StatusXml.xml"
TUNERSTATUS_URL = "/goform/formTuner_TunerXml.xml"
HDTUNERSTATUS_URL = "/goform/formTuner_HdXml.xml"
COMMAND_NETAUDIO_POST_URL = "/NetAudio/index.put.asp"
# Main Zone URLs
STATUS_URL = "/goform/formMainZone_MainZoneXmlStatus.xml"
MAINZONE_URL = "/goform/formMainZone_MainZoneXml.xml"
COMMAND_SEL_SRC_URL = "/goform/formiPhoneAppDirect.xml?SI"
COMMAND_FAV_SRC_URL = "/goform/formiPhoneAppDirect.xml?ZM"
COMMAND_POWER_ON_URL = "/goform/formiPhoneAppPower.xml?1+PowerOn"
COMMAND_POWER_STANDBY_URL = "/goform/formiPhoneAppPower.xml?1+PowerStandby"
COMMAND_VOLUME_UP_URL = "/goform/formiPhoneAppDirect.xml?MVUP"
COMMAND_VOLUME_DOWN_URL = "/goform/formiPhoneAppDirect.xml?MVDOWN"
COMMAND_SET_VOLUME_URL = "/goform/formiPhoneAppVolume.xml?1+%.1f"
COMMAND_MUTE_ON_URL = "/goform/formiPhoneAppMute.xml?1+MuteOn"
COMMAND_MUTE_OFF_URL = "/goform/formiPhoneAppMute.xml?1+MuteOff"
COMMAND_SEL_SM_URL = "/goform/formiPhoneAppDirect.xml?MS"
COMMAND_SET_ZST_URL = "/goform/formiPhoneAppDirect.xml?MN"
# Zone 2 URLs
STATUS_Z2_URL = "/goform/formZone2_Zone2XmlStatus.xml"
MAINZONE_Z2_URL = None
COMMAND_SEL_SRC_Z2_URL = "/goform/formiPhoneAppDirect.xml?Z2"
COMMAND_FAV_SRC_Z2_URL = "/goform/formiPhoneAppDirect.xml?Z2"
COMMAND_POWER_ON_Z2_URL = "/goform/formiPhoneAppPower.xml?2+PowerOn"
COMMAND_POWER_STANDBY_Z2_URL = "/goform/formiPhoneAppPower.xml?2+PowerStandby"
COMMAND_VOLUME_UP_Z2_URL = "/goform/formiPhoneAppDirect.xml?Z2UP"
COMMAND_VOLUME_DOWN_Z2_URL = "/goform/formiPhoneAppDirect.xml?Z2DOWN"
COMMAND_SET_VOLUME_Z2_URL = "/goform/formiPhoneAppVolume.xml?2+%.1f"
COMMAND_MUTE_ON_Z2_URL = "/goform/formiPhoneAppMute.xml?2+MuteOn"
COMMAND_MUTE_OFF_Z2_URL = "/goform/formiPhoneAppMute.xml?2+MuteOff"
# Zone 3 URLs
STATUS_Z3_URL = "/goform/formZone3_Zone3XmlStatus.xml"
MAINZONE_Z3_URL = None
COMMAND_SEL_SRC_Z3_URL = "/goform/formiPhoneAppDirect.xml?Z3"
COMMAND_FAV_SRC_Z3_URL = "/goform/formiPhoneAppDirect.xml?Z3"
COMMAND_POWER_ON_Z3_URL = "/goform/formiPhoneAppPower.xml?3+PowerOn"
COMMAND_POWER_STANDBY_Z3_URL = "/goform/formiPhoneAppPower.xml?3+PowerStandby"
COMMAND_VOLUME_UP_Z3_URL = "/goform/formiPhoneAppDirect.xml?Z3UP"
COMMAND_VOLUME_DOWN_Z3_URL = "/goform/formiPhoneAppDirect.xml?Z3DOWN"
COMMAND_SET_VOLUME_Z3_URL = "/goform/formiPhoneAppVolume.xml?3+%.1f"
COMMAND_MUTE_ON_Z3_URL = "/goform/formiPhoneAppMute.xml?3+MuteOn"
COMMAND_MUTE_OFF_Z3_URL = "/goform/formiPhoneAppMute.xml?3+MuteOff"
ReceiverURLs = namedtuple(
"ReceiverURLs", ["appcommand", "status", "mainzone", "deviceinfo",
"netaudiostatus", "tunerstatus", "hdtunerstatus",
"command_sel_src", "command_fav_src", "command_power_on",
"command_power_standby", "command_volume_up",
"command_volume_down", "command_set_volume",
"command_mute_on", "command_mute_off",
"command_sel_sound_mode", "command_netaudio_post",
"command_set_all_zone_stereo"])
DENONAVR_URLS = ReceiverURLs(appcommand=APPCOMMAND_URL,
status=STATUS_URL,
mainzone=MAINZONE_URL,
deviceinfo=DEVICEINFO_URL,
netaudiostatus=NETAUDIOSTATUS_URL,
tunerstatus=TUNERSTATUS_URL,
hdtunerstatus=HDTUNERSTATUS_URL,
command_sel_src=COMMAND_SEL_SRC_URL,
command_fav_src=COMMAND_FAV_SRC_URL,
command_power_on=COMMAND_POWER_ON_URL,
command_power_standby=COMMAND_POWER_STANDBY_URL,
command_volume_up=COMMAND_VOLUME_UP_URL,
command_volume_down=COMMAND_VOLUME_DOWN_URL,
command_set_volume=COMMAND_SET_VOLUME_URL,
command_mute_on=COMMAND_MUTE_ON_URL,
command_mute_off=COMMAND_MUTE_OFF_URL,
command_sel_sound_mode=COMMAND_SEL_SM_URL,
command_netaudio_post=COMMAND_NETAUDIO_POST_URL,
command_set_all_zone_stereo=COMMAND_SET_ZST_URL)
ZONE2_URLS = ReceiverURLs(appcommand=APPCOMMAND_URL,
status=STATUS_Z2_URL,
mainzone=MAINZONE_Z2_URL,
deviceinfo=DEVICEINFO_URL,
netaudiostatus=NETAUDIOSTATUS_URL,
tunerstatus=TUNERSTATUS_URL,
hdtunerstatus=HDTUNERSTATUS_URL,
command_sel_src=COMMAND_SEL_SRC_Z2_URL,
command_fav_src=COMMAND_FAV_SRC_Z2_URL,
command_power_on=COMMAND_POWER_ON_Z2_URL,
command_power_standby=COMMAND_POWER_STANDBY_Z2_URL,
command_volume_up=COMMAND_VOLUME_UP_Z2_URL,
command_volume_down=COMMAND_VOLUME_DOWN_Z2_URL,
command_set_volume=COMMAND_SET_VOLUME_Z2_URL,
command_mute_on=COMMAND_MUTE_ON_Z2_URL,
command_mute_off=COMMAND_MUTE_OFF_Z2_URL,
command_sel_sound_mode=COMMAND_SEL_SM_URL,
command_netaudio_post=COMMAND_NETAUDIO_POST_URL,
command_set_all_zone_stereo=COMMAND_SET_ZST_URL)
ZONE3_URLS = ReceiverURLs(appcommand=APPCOMMAND_URL,
status=STATUS_Z3_URL,
mainzone=MAINZONE_Z3_URL,
deviceinfo=DEVICEINFO_URL,
netaudiostatus=NETAUDIOSTATUS_URL,
tunerstatus=TUNERSTATUS_URL,
hdtunerstatus=HDTUNERSTATUS_URL,
command_sel_src=COMMAND_SEL_SRC_Z3_URL,
command_fav_src=COMMAND_FAV_SRC_Z3_URL,
command_power_on=COMMAND_POWER_ON_Z3_URL,
command_power_standby=COMMAND_POWER_STANDBY_Z3_URL,
command_volume_up=COMMAND_VOLUME_UP_Z3_URL,
command_volume_down=COMMAND_VOLUME_DOWN_Z3_URL,
command_set_volume=COMMAND_SET_VOLUME_Z3_URL,
command_mute_on=COMMAND_MUTE_ON_Z3_URL,
command_mute_off=COMMAND_MUTE_OFF_Z3_URL,
command_sel_sound_mode=COMMAND_SEL_SM_URL,
command_netaudio_post=COMMAND_NETAUDIO_POST_URL,
command_set_all_zone_stereo=COMMAND_SET_ZST_URL)
POWER_ON = "ON"
POWER_OFF = "OFF"
POWER_STANDBY = "STANDBY"
STATE_ON = "on"
STATE_OFF = "off"
STATE_PLAYING = "playing"
STATE_PAUSED = "paused"
NO_ZONES = None
ZONE2 = {"Zone2": None}
ZONE3 = {"Zone3": None}
ZONE2_ZONE3 = {"Zone2": None, "Zone3": None}
class DenonAVR:
"""Representing a Denon AVR Device."""
def __init__(self, host, name=None, show_all_inputs=False, timeout=2.0,
add_zones=NO_ZONES):
"""
Initialize MainZone of DenonAVR.
:param host: IP or HOSTNAME.
:type host: str
:param name: Device name, if None FriendlyName of device is used.
:type name: str or None
:param show_all_inputs: If True deleted input functions are also shown
:type show_all_inputs: bool
:param add_zones: Additional Zones for which an instance are created
:type add_zones: dict [str, str] or None
"""
self._name = name
self._host = host
# Main zone just set for DenonAVR class
if self.__class__.__name__ == "DenonAVR":
self._zone = "Main"
self._zones = {self._zone: self}
if self._zone == "Main":
self._urls = DENONAVR_URLS
elif self._zone == "Zone2":
self._urls = ZONE2_URLS
elif self._zone == "Zone3":
self._urls = ZONE3_URLS
else:
raise ValueError("Invalid zone {}".format(self._zone))
# Timeout for HTTP calls to receiver
self.timeout = timeout
# Receiver types could be avr, avr-x, avr-x-2016 after being determined
self._receiver_type = None
# Port 80 for avr and avr-x, Port 8080 port avr-x-2016
self._receiver_port = None
self._show_all_inputs = show_all_inputs
self._mute = STATE_OFF
self._volume = "--"
self._input_func = None
self._input_func_list = {}
self._input_func_list_rev = {}
self._sound_mode_raw = None
self._sound_mode_dict = SOUND_MODE_MAPPING
self._support_sound_mode = None
self._sm_match_dict = self.construct_sm_match_dict()
self._netaudio_func_list = []
self._playing_func_list = []
self._favorite_func_list = []
self._state = None
self._power = None
self._image_url = None
self._image_available = None
self._title = None
self._artist = None
self._album = None
self._band = None
self._frequency = None
self._station = None
# Determine receiver type and input functions
self._update_input_func_list()
if self._receiver_type == AVR_X_2016.type:
self._get_zone_name()
else:
self._get_receiver_name()
# Determine if sound mode is supported
self._get_support_sound_mode()
# Get initial setting of values
self.update()
# Create instances of additional zones if requested
if self._zone == "Main" and add_zones is not None:
self.create_zones(add_zones)
def exec_appcommand_post(self, attribute_list):
"""
Prepare and execute a HTTP POST call to AppCommand.xml end point.
Returns XML ElementTree on success and None on fail.
"""
# Prepare POST XML body for AppCommand.xml
post_root = ET.Element("tx")
for attribute in attribute_list:
# Append tags for each attribute
item = ET.Element("cmd")
item.set("id", "1")
item.text = attribute
post_root.append(item)
# Buffer XML body as binary IO
body = BytesIO()
post_tree = ET.ElementTree(post_root)
post_tree.write(body, encoding="utf-8", xml_declaration=True)
# Query receivers AppCommand.xml
try:
res = self.send_post_command(
self._urls.appcommand, body.getvalue())
except requests.exceptions.RequestException:
_LOGGER.error("No connection to %s end point on host %s",
self._urls.appcommand, self._host)
body.close()
else:
# Buffered XML not needed anymore: close
body.close()
try:
# Return XML ElementTree
root = ET.fromstring(res)
except (ET.ParseError, TypeError):
_LOGGER.error(
"End point %s on host %s returned malformed XML.",
self._urls.appcommand, self._host)
else:
return root
def get_status_xml(self, command, suppress_errors=False):
"""Get status XML via HTTP and return it as XML ElementTree."""
# Get XML structure via HTTP get
res = requests.get("http://{host}:{port}{command}".format(
host=self._host, port=self._receiver_port, command=command),
timeout=self.timeout)
# Continue with XML processing only if HTTP status code = 200
if res.status_code == 200:
try:
# Return XML ElementTree
return ET.fromstring(res.text)
except ET.ParseError as err:
if not suppress_errors:
_LOGGER.error(
"Host %s returned malformed XML for end point %s",
self._host, command)
_LOGGER.error(err)
raise ValueError
else:
if not suppress_errors:
_LOGGER.error((
"Host %s returned HTTP status code %s to GET request at "
"end point %s"), self._host, res.status_code, command)
raise ValueError
def send_get_command(self, command):
"""Send command via HTTP get to receiver."""
# Send commands via HTTP get
res = requests.get("http://{host}:{port}{command}".format(
host=self._host, port=self._receiver_port, command=command),
timeout=self.timeout)
if res.status_code == 200:
return True
else:
_LOGGER.error((
"Host %s returned HTTP status code %s to GET command at "
"end point %s"), self._host, res.status_code, command)
return False
def send_post_command(self, command, body):
"""Send command via HTTP post to receiver."""
# Send commands via HTTP post
res = requests.post("http://{host}:{port}{command}".format(
host=self._host, port=self._receiver_port, command=command),
data=body, timeout=self.timeout)
if res.status_code == 200:
return res.text
else:
_LOGGER.error((
"Host %s returned HTTP status code %s to POST command at "
"end point %s"), self._host, res.status_code, command)
return False
def create_zones(self, add_zones):
"""Create instances of additional zones for the receiver."""
for zone, zname in add_zones.items():
# Name either set explicitly or name of Main Zone with suffix
zonename = "{} {}".format(self._name, zone) if (
zname is None) else zname
zone_inst = DenonAVRZones(self, zone, zonename)
self._zones[zone] = zone_inst
def update(self):
"""
Get the latest status information from device.
Method executes the update method for the current receiver type.
"""
if self._receiver_type == AVR_X_2016.type:
return self._update_avr_2016()
else:
return self._update_avr()
def _update_avr(self):
"""
Get the latest status information from device.
Method queries device via HTTP and updates instance attributes.
Returns "True" on success and "False" on fail.
This method is for pre 2016 AVR(-X) devices
"""
# Set all tags to be evaluated
relevant_tags = {"Power": None, "InputFuncSelect": None, "Mute": None,
"MasterVolume": None}
# Sound mode information only available in main zone
if self._zone == "Main" and self._support_sound_mode:
relevant_tags["selectSurround"] = None
relevant_tags["SurrMode"] = None
# Get status XML from Denon receiver via HTTP
try:
root = self.get_status_xml(self._urls.status)
except ValueError:
pass
except requests.exceptions.RequestException:
# On timeout and connection error, the device is probably off
self._power = POWER_OFF
else:
# Get the tags from this XML
relevant_tags = self._get_status_from_xml_tags(root, relevant_tags)
# Second option to update variables from different source
if relevant_tags and self._power != POWER_OFF:
try:
root = self.get_status_xml(self._urls.mainzone)
except (ValueError,
requests.exceptions.RequestException):
pass
else:
# Get the tags from this XML
relevant_tags = self._get_status_from_xml_tags(root,
relevant_tags)
# Error message if still some variables are not updated yet
if relevant_tags and self._power != POWER_OFF:
_LOGGER.error("Missing status information from XML of %s for: %s",
self._zone, ", ".join(relevant_tags.keys()))
# Set state and media image URL based on current source
# and power status
if (self._power == POWER_ON) and (
self._input_func in self._playing_func_list):
if self._update_media_data():
pass
else:
_LOGGER.error(
"Update of media data for source %s in %s failed",
self._input_func, self._zone)
elif self._power == POWER_ON:
self._state = STATE_ON
self._title = None
self._artist = None
self._album = None
self._band = None
self._frequency = None
self._station = None
self._image_url = None
else:
self._state = STATE_OFF
self._title = None
self._artist = None
self._album = None
self._band = None
self._frequency = None
self._station = None
# Get/update sources list if current source is not known yet
if (self._input_func not in self._input_func_list and
self._input_func is not None):
if self._update_input_func_list():
_LOGGER.info("List of input functions refreshed.")
# If input function is still not known, create new entry.
if (self._input_func not in self._input_func_list and
self._input_func is not None):
inputfunc = self._input_func
self._input_func_list_rev[inputfunc] = inputfunc
self._input_func_list[inputfunc] = inputfunc
else:
_LOGGER.error((
"Input function list for Denon receiver at host %s "
"could not be updated."), self._host)
# Finished
return True
def _update_avr_2016(self):
"""
Get the latest status information from device.
Method queries device via HTTP and updates instance attributes.
Returns "True" on success and "False" on fail.
This method is for AVR-X devices built in 2016 and later.
"""
# Collect tags for AppCommand.xml call
tags = ["GetAllZonePowerStatus", "GetAllZoneSource",
"GetAllZoneVolume", "GetAllZoneMuteStatus",
"GetSurroundModeStatus"]
# Execute call
root = self.exec_appcommand_post(tags)
# Check result
if root is None:
_LOGGER.error("Update failed.")
return False
# Extract relevant information
zone = self._get_own_zone()
try:
self._power = root[0].find(zone).text
except (AttributeError, IndexError):
_LOGGER.error("No PowerStatus found for zone %s", self.zone)
try:
self._mute = root[3].find(zone).text
except (AttributeError, IndexError):
_LOGGER.error("No MuteStatus found for zone %s", self.zone)
try:
self._volume = root.find(
"./cmd/{zone}/volume".format(zone=zone)).text
except AttributeError:
_LOGGER.error("No VolumeStatus found for zone %s", self.zone)
try:
inputfunc = root.find(
"./cmd/{zone}/source".format(zone=zone)).text
except AttributeError:
_LOGGER.error("No Source found for zone %s", self.zone)
else:
try:
self._input_func = self._input_func_list_rev[inputfunc]
except KeyError:
_LOGGER.info("No mapping for source %s found", inputfunc)
self._input_func = inputfunc
# Get/update sources list if current source is not known yet
if self._update_input_func_list():
_LOGGER.info("List of input functions refreshed.")
# If input function is still not known, create new entry.
if (inputfunc not in self._input_func_list and
inputfunc is not None):
self._input_func_list_rev[inputfunc] = inputfunc
self._input_func_list[inputfunc] = inputfunc
else:
_LOGGER.error((
"Input function list for Denon receiver at host %s "
"could not be updated."), self._host)
try:
self._sound_mode_raw = root[4][0].text.rstrip()
except (AttributeError, IndexError):
_LOGGER.error("No SoundMode found for the main zone %s", self.zone)
# Now playing information is not implemented for 2016+ models, because
# a HEOS API query needed. So only sync the power state for now.
if self._power == POWER_ON:
self._state = STATE_ON
else:
self._state = STATE_OFF
return True
def _update_input_func_list(self):
"""
Update sources list from receiver.
Internal method which updates sources list of receiver after getting
sources and potential renaming information from receiver.
"""
# Get all sources and renaming information from receiver
# For structural information of the variables please see the methods
receiver_sources = self._get_receiver_sources()
if not receiver_sources:
_LOGGER.error("Receiver sources list empty. "
"Please check if device is powered on.")
return False
# First input_func_list determination of AVR-X receivers
if self._receiver_type in [AVR_X.type, AVR_X_2016.type]:
renamed_sources, deleted_sources, status_success = (
self._get_renamed_deleted_sourcesapp())
# Backup if previous try with AppCommand was not successful
if not status_success:
renamed_sources, deleted_sources = (
self._get_renamed_deleted_sources())
# Remove all deleted sources
if self._show_all_inputs is False:
for deleted_source in deleted_sources.items():
if deleted_source[1] == "DEL":
receiver_sources.pop(deleted_source[0], None)
# Clear and rebuild the sources lists
self._input_func_list.clear()
self._input_func_list_rev.clear()
self._netaudio_func_list.clear()
self._playing_func_list.clear()
for item in receiver_sources.items():
# Mapping of item[0] because some func names are inconsistant
# at AVR-X receivers
m_item_0 = SOURCE_MAPPING.get(item[0], item[0])
# For renamed sources use those names and save the default name
# for a later mapping
if item[0] in renamed_sources:
self._input_func_list[renamed_sources[item[0]]] = m_item_0
self._input_func_list_rev[
m_item_0] = renamed_sources[item[0]]
# If the source is a netaudio source, save its renamed name
if item[0] in NETAUDIO_SOURCES:
self._netaudio_func_list.append(
renamed_sources[item[0]])
# If the source is a playing source, save its renamed name
if item[0] in PLAYING_SOURCES:
self._playing_func_list.append(
renamed_sources[item[0]])
# Otherwise the default names are used
else:
self._input_func_list[item[1]] = m_item_0
self._input_func_list_rev[m_item_0] = item[1]
# If the source is a netaudio source, save its name
if item[1] in NETAUDIO_SOURCES:
self._netaudio_func_list.append(item[1])
# If the source is a playing source, save its name
if item[1] in PLAYING_SOURCES:
self._playing_func_list.append(item[1])
# Determination of input_func_list for non AVR-nonX receivers
elif self._receiver_type == AVR.type:
# Clear and rebuild the sources lists
self._input_func_list.clear()
self._input_func_list_rev.clear()
self._netaudio_func_list.clear()
self._playing_func_list.clear()
for item in receiver_sources.items():
self._input_func_list[item[1]] = item[0]
self._input_func_list_rev[item[0]] = item[1]
# If the source is a netaudio source, save its name
if item[0] in NETAUDIO_SOURCES:
self._netaudio_func_list.append(item[1])
# If the source is a playing source, save its name
if item[0] in PLAYING_SOURCES:
self._playing_func_list.append(item[1])
else:
_LOGGER.error('Receiver type not set yet.')
return False
# Finished
return True
def _get_receiver_name(self):
"""Get name of receiver from web interface if not set."""
# If name is not set yet, get it from Main Zone URL
if self._name is None and self._urls.mainzone is not None:
name_tag = {"FriendlyName": None}
try:
root = self.get_status_xml(self._urls.mainzone)
except (ValueError,
requests.exceptions.RequestException):
_LOGGER.warning("Receiver name could not be determined. "
"Using standard name: Denon AVR.")
self._name = "Denon AVR"
else:
# Get the tags from this XML
name_tag = self._get_status_from_xml_tags(root, name_tag)
if name_tag:
_LOGGER.warning("Receiver name could not be determined. "
"Using standard name: Denon AVR.")
self._name = "Denon AVR"
def _get_zone_name(self):
"""Get receivers zone name if not set yet."""
if self._name is None:
# Collect tags for AppCommand.xml call
tags = ["GetZoneName"]
# Execute call
root = self.exec_appcommand_post(tags)
# Check result
if root is None:
_LOGGER.error("Getting ZoneName failed.")
else:
zone = self._get_own_zone()
try:
name = root.find(
"./cmd/{zone}".format(zone=zone)).text
except AttributeError:
_LOGGER.error("No ZoneName found for zone %s", self.zone)
else:
self._name = name.strip()
def _get_support_sound_mode(self):
"""
Get if sound mode is supported from device.
Method executes the method for the current receiver type.
"""
if self._receiver_type == AVR_X_2016.type:
return self._get_support_sound_mode_avr_2016()
else:
return self._get_support_sound_mode_avr()
def _get_support_sound_mode_avr(self):
"""
Get if sound mode is supported from device.
Method queries device via HTTP.
Returns "True" if sound mode supported and "False" if not.
This method is for pre 2016 AVR(-X) devices
"""
# Set sound mode tags to be checked if available
relevant_tags = {"selectSurround": None, "SurrMode": None}
# Get status XML from Denon receiver via HTTP
try:
root = self.get_status_xml(self._urls.status)
except (ValueError, requests.exceptions.RequestException):
pass
else:
# Process the tags from this XML
relevant_tags = self._get_status_from_xml_tags(root, relevant_tags)
# Second option to update variables from different source
if relevant_tags:
try:
root = self.get_status_xml(self._urls.mainzone)
except (ValueError,
requests.exceptions.RequestException):
pass
else:
# Get the tags from this XML
relevant_tags = self._get_status_from_xml_tags(root,
relevant_tags)
# if sound mode not found in the status XML, return False
if relevant_tags:
self._support_sound_mode = False
return False
# if sound mode found, the relevant_tags are empty: return True.
self._support_sound_mode = True
return True
def _get_support_sound_mode_avr_2016(self):
"""
Get if sound mode is supported from device.
Method enables sound mode.
Returns "True" in all cases for 2016 AVR(-X) devices
"""
self._support_sound_mode = True
return True
def _get_renamed_deleted_sources(self):
"""
Get renamed and deleted sources lists from receiver .
Internal method which queries device via HTTP to get names of renamed
input sources.
"""
# renamed_sources and deleted_sources are dicts with "source" as key
# and "renamed_source" or deletion flag as value.
renamed_sources = {}
deleted_sources = {}
xml_inputfunclist = []
xml_renamesource = []
xml_deletesource = []
# This XML is needed to get names of eventually renamed sources
try:
# AVR-X and AVR-nonX using different XMLs to provide info about
# deleted sources
if self._receiver_type == AVR_X.type:
root = self.get_status_xml(self._urls.status)
# URL only available for Main Zone.
elif self._receiver_type == AVR.type:
if self._urls.mainzone is not None:
root = self.get_status_xml(self._urls.mainzone)
else:
root = self.get_status_xml(self._urls.status)
else:
return (renamed_sources, deleted_sources)
except (ValueError, requests.exceptions.RequestException):
return (renamed_sources, deleted_sources)
# Get the relevant tags from XML structure
for child in root:
# Default names of the sources
if child.tag == "InputFuncList":
for value in child:
xml_inputfunclist.append(value.text)
# Renamed sources
if child.tag == "RenameSource":
for value in child:
# Two different kinds of source structure types exist
# 1. <RenameSource><Value>...
if value.text is not None:
xml_renamesource.append(value.text.strip())
# 2. <RenameSource><Value><Value>
else:
try:
xml_renamesource.append(value[0].text.strip())
# Exception covers empty tags and appends empty line
# in this case, to ensure that sources and
# renamed_sources lists have always the same length
except IndexError:
xml_renamesource.append(None)
# Deleted sources
if child.tag == "SourceDelete":
for value in child:
xml_deletesource.append(value.text)
# If the deleted source list is empty then use all sources.
if not xml_deletesource:
xml_deletesource = ['USE'] * len(xml_inputfunclist)
# Renamed and deleted sources are in the same row as the default ones
# Only values which are not None are considered. Otherwise translation
# is not valid and original name is taken
for i, item in enumerate(xml_inputfunclist):
try:
if xml_renamesource[i] is not None:
renamed_sources[item] = xml_renamesource[i]
else:
renamed_sources[item] = item
except IndexError:
_LOGGER.error(
"List of renamed sources incomplete, continuing anyway")
try:
deleted_sources[item] = xml_deletesource[i]
except IndexError:
_LOGGER.error(
"List of deleted sources incomplete, continuing anyway")
return (renamed_sources, deleted_sources)
def _get_renamed_deleted_sourcesapp(self):
"""
Get renamed and deleted sources lists from receiver .
Internal method which queries device via HTTP to get names of renamed
input sources. In this method AppCommand.xml is used.
"""
# renamed_sources and deleted_sources are dicts with "source" as key
# and "renamed_source" or deletion flag as value.
renamed_sources = {}
deleted_sources = {}
# Collect tags for AppCommand.xml call
tags = ["GetRenameSource", "GetDeletedSource"]
# Execute call
root = self.exec_appcommand_post(tags)
# Check result
if root is None:
_LOGGER.error("Getting renamed and deleted sources failed.")
return (renamed_sources, deleted_sources, False)
# Detect "Document Error: Data follows" title if URL does not exist
document_error = root.find("./head/title")
if document_error is not None:
if document_error.text == "Document Error: Data follows":
return (renamed_sources, deleted_sources, False)
for child in root.findall("./cmd/functionrename/list"):
try:
renamed_sources[child.find("name").text.strip()] = (
child.find("rename").text.strip())
except AttributeError:
continue
for child in root.findall("./cmd/functiondelete/list"):
try:
deleted_sources[child.find("FuncName").text.strip(
)] = "DEL" if (
child.find("use").text.strip() == "0") else None
except AttributeError:
continue
return (renamed_sources, deleted_sources, True)
def _get_receiver_sources(self):
"""
Get sources list from receiver.
Internal method which queries device via HTTP to get the receiver's
input sources.
This method also determines the type of the receiver
(avr, avr-x, avr-x-2016).
"""
# Test if receiver is a AVR-X with port 80 for pre 2016 devices and
# port 8080 devices 2016 and later
r_types = [AVR_X, AVR_X_2016]
for r_type, port in r_types:
self._receiver_port = port
# This XML is needed to get the sources of the receiver
try:
root = self.get_status_xml(self._urls.deviceinfo,
suppress_errors=True)
except (ValueError, requests.exceptions.RequestException):
self._receiver_type = None
else:
# First test by CommApiVers
try:
if bool(DEVICEINFO_COMMAPI_PATTERN.search(
root.find("CommApiVers").text) is not None):
self._receiver_type = r_type
# receiver found break the loop
break
except AttributeError:
# AttributeError occurs when ModelName tag is not found.
# In this case there is no AVR-X device
self._receiver_type = None
# if first test did not find AVR-X device, check by model name
if self._receiver_type is None:
try:
if bool(DEVICEINFO_AVR_X_PATTERN.search(
root.find("ModelName").text) is not None):
self._receiver_type = r_type
# receiver found break the loop
break
except AttributeError:
# AttributeError occurs when ModelName tag is not found
# In this case there is no AVR-X device
self._receiver_type = None
# Set ports and update method
if self._receiver_type is None:
self._receiver_type = AVR.type
self._receiver_port = AVR.port
elif self._receiver_type == AVR_X_2016.type:
self._receiver_port = AVR_X_2016.port
else:
self._receiver_port = AVR_X.port
_LOGGER.info("Identified receiver type: '%s' on port: '%s'",
self._receiver_type, self._receiver_port)
# Not an AVR-X device, start determination of sources
if self._receiver_type == AVR.type:
# Sources list is equal to list of renamed sources.
non_x_sources, deleted_non_x_sources, status_success = (
self._get_renamed_deleted_sourcesapp())
# Backup if previous try with AppCommand was not successful
if not status_success:
non_x_sources, deleted_non_x_sources = (
self._get_renamed_deleted_sources())
# Remove all deleted sources
if self._show_all_inputs is False:
for deleted_source in deleted_non_x_sources.items():
if deleted_source[1] == "DEL":
non_x_sources.pop(deleted_source[0], None)
# Invalid source "SOURCE" needs to be deleted
non_x_sources.pop("SOURCE", None)
return non_x_sources
# Following source determination of AVR-X receivers
else:
# receiver_sources is of type dict with "FuncName" as key and
# "DefaultName" as value.
receiver_sources = {}
# Source determination from XML
favorites = root.find(".//FavoriteStation")
if favorites:
for child in favorites:
if not child.tag.startswith("Favorite"):
continue
func_name = child.tag.upper()
self._favorite_func_list.append(func_name)
receiver_sources[func_name] = child.find("Name").text
for xml_zonecapa in root.findall("DeviceZoneCapabilities"):
# Currently only Main Zone (No=0) supported