CODEWITHSUNDEEP
pythonpython-3.x
user1209325
user1209325

Reputation: 83

Best way to communicate with telnet api from python?

I have a multi-room speaker system from Denon called Heos which I want to control by using python script. To communicate with the multi-room system I have to telnet to port 1255 on the device and send commands like this:

heos://player/set_play_state?pid=player_id&state=play_state

The response back is in json:

{
 "heos": {
 "command": " player/set_play_state ",
 "result": "success",
 "message": "pid='player_id'&state='play_state'"
 }
}

I have successfully used python telnet lib to send simple commands like this:

command = "heos://player/set_play_state?pid=player_id&state=play_state"
telnet.write(command.encode('ASCII') + b'\r\n')

But what is the best way to get the response back in a usable format? Loop with telnet.read_until? I want to result and message lines back to a clean variable.

This method with using telnet to communicate with api feels a bit dirty. Is it possible to use something else, for example socket?

Thanks in advance

The API/CLI is documented here: http://rn.dmglobal.com/euheos/HEOS_CLI_ProtocolSpecification.pdf

Upvotes: 0

Views: 1983

Answers (2)

Stephan
Stephan

Reputation: 392

FWIW, here is my GitHub repo (inspired by this repo) for controlling a HEOS speaker with Python. It uses a similar approach as the accepted result, but additionally waits if the HEOS player is busy.

def telnet_request(self, command, wait = True):
    """Execute a `command` and return the response(s)."""
    command = self.heosurl + command
    logging.debug("telnet request {}".format(command))
    self.telnet.write(command.encode('ASCII') + b'\r\n')
    response = b''
    while True:
        response += self.telnet.read_some()
        try:
            response = json.loads(response)
            if not wait:
                logging.debug("I accept the first response: {}".format(response))
                break
            # sometimes, I get a response with the message "under
            # process". I might want to wait here
            message = response.get("heos", {}).get("message", "")
            if "command under process" not in message:
                logging.debug("I assume this is the final response: {}".format(response))
                break
            logging.debug("Wait for the final response")
            response = b'' # forget this message
        except ValueError:
            # response is not a complete JSON object
            pass
        except TypeError:
            # response is not a complete JSON object
            pass

    if response.get("result") == "fail":
        logging.warn(response)
        return None

    return response

Upvotes: 0

glibdud
glibdud

Reputation: 7880

While it may be possible to use loop_until() here, it would depend on exactly how the response JSON is formatted, and it would probably be unwise to rely on it.

If the remote device closes the connection after sending the response, the easy way would be a simple

response = json.loads(telnet.read_all().decode())

If it remains open for more commands, then you'll instead need to keep receiving until you have a complete JSON object. Here's a possibility that just keeps trying to parse the JSON until it succeeds:

response = ''
while True:
    response += telnet.read_some().decode()
    try:
        response = json.loads(response)
        break
    except ValueError:
        pass

Either way, your result and message are response['heos']['result'] and response['heos']['message'].

Upvotes: 1

Related Questions