pytheos.networking Package

discovery Module

This module provides the discovery functionality for Pytheos

async pytheos.networking.discovery.discover(timeout: int) → List[pytheos.networking.types.SSDPResponse]

Convenience function for initiating the discovery process.

Parameters

timeout – Optional override for the default timeout

Returns

list

class pytheos.networking.discovery.SSDPBroadcastMessage(address: str, port: int, service: str, mx: int)

Bases: object

Representation of an SSDP message

class pytheos.networking.discovery.SSDPProtocol(address, bind_ip, bind_port, reuse_addr, ttl)

Bases: asyncio.protocols.DatagramProtocol

connection_made(transport: asyncio.transports.BaseTransport)

Initializes the socket given the provided transport.

Parameters

transport – Transport

Returns

None

datagram_received(data: bytes, addr: Tuple[str, int])

Callback used to add response datagrams to our results.

Parameters

  • data – Datagram bytes

  • addr – Source address

Returns

None

get_results()

Retrieves the broadcast results

Returns

list of results

class pytheos.networking.discovery.Discovery(broadcast_address: Optional[str] = '239.255.255.250', broadcast_port: Optional[int] = 1900, bind_ip: Optional[str] = None)

Bases: object

HEOS discovery implementation

DEFAULT_BROADCAST_ADDRESS = '239.255.255.250'
DEFAULT_BROADCAST_PORT = 1900
DEFAULT_BIND_PORT = 12112
DEFAULT_TTL = 3
DEFAULT_SERVICE = 'urn:schemas-denon-com:device:ACT-Denon:1'
DEFAULT_TIMEOUT = 5
DEFAULT_RETRIES = 1
DEFAULT_MX = 3
async discover(timeout)

Sends a discovery broadcast to the network and waits for responses.

Parameters

timeout – Amount of time to wait for a response.

Returns

list of results

connection Module

Provides the implementation for the Connection class

class pytheos.networking.connection.Connection

Bases: object

Connection to the telnet service on a HEOS device

CONNECTION_READ_TIMEOUT = 1
MESSAGE_READ_TIMEOUT = 1
DELAY_MESSAGES = ('command under process', 'processing previous command')
property connected
property prettify_json_response
async connect(server: str, port: int, deduplicate: bool = False)

Establish a connection with the HEOS service

Parameters

  • server – Server hostname or IP

  • port – Port number

  • deduplicate – Flag to enable message deduplication

Returns

None

write(input_data: bytes)

Writes the provided data to the connection

Parameters

input_data – Data to write

Returns

None

async read_until(target: bytes) → bytes

Reads from the connection until the target string is found or the optional timeout is hit

Parameters

target – Target string

Returns

str

async call(group: str, command: str, **kwargs: dict) → pytheos.models.heos.HEOSResult

Formats a HEOS API request, submits it, and reads the response.

Parameters

  • group – Group name (e.g. system, player, etc)

  • command – Command name (e.g. heart_beat)

  • kwargs – Any parameters that should be sent along with the command

Raises

AssertionError, CommandFailedError

Returns

HEOSResult

send_command(group: str, command: str, **kwargs: dict) → None

Formats a HEOS API request and submits it

Parameters

  • group – Group name (e.g. system, player, etc)

  • command – Command name (e.g. heart_beat)

  • kwargs – Any parameters that should be sent along with the command

Raises

AssertionError, CommandFailedError

Returns

HEOSResult

async read_message(timeout: int = 1, delimiter: bytes = b'\r\n') → Optional[dict]

Reads a message from the connection

Parameters

  • timeout – Timeout (seconds)

  • delimiter – Message delimiter

Returns

bytes

async heart_beat()

Performs a system/heart_beat API call on the connection in order to keep the connection alive.

Returns

None

errors Module

class pytheos.networking.errors.HEOSErrorCode(value)

Bases: enum.Enum

An enumeration.

UnrecognizedCommand = 1
InvalidID = 2
WrongNumberOfCommandArguments = 3
RequestedDataNotAvailable = 4
ResourceCurrentlyNotAvailable = 5
InvalidCredentials = 6
CommandCouldNotBeExecuted = 7
UserNotLoggedIn = 8
ParameterOutOfRange = 9
UserNotFound = 10
InternalError = 11
SystemError = 12
ProcessingPreviousCommand = 13
MediaCannotBePlayed = 14
OptionNotSupported = 15
CommandQueueFull = 16
ReachedSkipLimit = 17
class pytheos.networking.errors.HEOSSystemErrorCode(value)

Bases: enum.Enum

An enumeration.

RemoteServiceReturnedError = -9
UserNotRegisters = -1061
UserNotLoggedIn = -1063
UserNotFound = -1056
AuthenticationError = -1201
AuthorizationError = -1232
UserParametersInvalid = -1239
exception pytheos.networking.errors.ChannelUnavailableError

Bases: pytheos.errors.PytheosError

Error returned when a channel is unavailable

exception pytheos.networking.errors.CommandFailedError(message: str, result: Optional[pytheos.models.heos.HEOSResult])

Bases: pytheos.errors.PytheosError

Error returned with a command fails to execute

exception pytheos.networking.errors.SignInFailedError(message: str, result: Optional[pytheos.models.heos.HEOSResult])

Bases: pytheos.networking.errors.CommandFailedError

Error returned when the system/sign_in command fails

exception pytheos.networking.errors.InvalidResponse

Bases: pytheos.errors.PytheosError

Error returned when the response to a command appears invalid

types Module

class pytheos.networking.types.SSDPResponse(response)

Bases: object