Telnet commands with pyscript ?

[vincegre]

Hi

I'm just starting with pyscript and Python too :D I need to control a device by connecting through telnet to it and sending a command with 2 parameters (input and output).
I have seen I can get it show like a service in HA so I started it like that but is there a way to do a telnet in pyscript ?

@service
def commutation(input=None, output=None):
    """hello_world example using pyscript."""
    log.info(f"service started")

Thanks for suggestions

Vincèn

[craigbarratt]

Good options are to use asynctelnet or telnetlib3, since those are non-blocking async implementation. You could start with a standalone program to make sure it works before trying it with pyscript.

An alternative is to use the older standard (now deprecated) telnetlib, but it blocks on I/O. So the only way to safely use that in pyscript is by running the telnet code in its own thread via task_executor() . See the docs if you take this path.

[vincegre]

Thanks for your good advices but unhappy pyscript is unable to import telnetlib3 :(

Code:

@service
def commutation(input=None, output=None):
    """blustream commutation."""
    log.info(f"service started")
    import asyncio, telnetlib3

    async def shell(reader, writer):
        while True:
            outp = await reader.read(1024)
            writer.write('OUT{output}FR{input}\r\n')

            # display all server output
            # print(outp, flush=True)

        # EOF
        print()

    loop = asyncio.get_event_loop()
    coro = telnetlib3.open_connection('localhost', 23, shell=shell)
    reader, writer = loop.run_until_complete(coro)
    loop.run_until_complete(writer.protocol.waiter_closed)

HA logs when I try to run it:
Exception in <file.blustream.commutation> line 5: import asyncio, telnetlib3 ^ ModuleNotFoundError: No module named 'telnetlib3'

Did I miss something ?

[craigbarratt]

It looks like you need to install telnetlib3.

On your code, you shouldn't be manipulating the event loop directly - pyscript handles that for you. (For a standalone program you do need to run the co-routine using the event loop as you have done.)

If you just want to write a client to connect to a tcp port, it might be simplest to just use a tcp socket. The documentation has this example:

import asyncio

Reader, Writer = None, None

@time_trigger('startup')
def do_client_connection():
    global Reader, Writer
    Reader, Writer = asyncio.open_connection('127.0.0.1', 23)

def client_send(message):
    if not Writer:
       raise("Client is not connected")
    Writer.write(message.encode())
    Writer.drain()
    return Reader.readline().decode()

This initiates the connection at startup, and provides function you can call that writes and reads from the socket. If you want to continually listen to the socket, you can do that in the startup function:

import asyncio

@time_trigger('startup')
def client_connect_and_listen():
    reader, writer = asyncio.open_connection('127.0.0.1', 23)

    while True:
        outp = await reader.read(1024)
        writer.write('OUT{output}FR{input}\r\n')
        # display all server output
        # print(outp, flush=True)

[vincegre]

On your code, you shouldn't be manipulating the event loop directly - pyscript handles that for you. (For a standalone program you do need to run the co-routine using the event loop as you have done.)

If you just want to write a client to connect to a tcp port, it might be simplest to just use a tcp socket. The documentation has this example:

Thanks a lot for your help as I'm really a beginner both with pyscript and python so not easy :-D

Using your example I think i got something nearly good:

@service
def commutation(input=None, output=None):
    log.info("service started")
    import asyncio
    Reader, Writer = None, None
    log.info("telnet starting")
    Reader, Writer = asyncio.open_connection('172.16.2.11', 50005)
    if not Writer:
       raise("Client is not connected")
    Writer.write('OUT{output}FR{input}\r\n')
    Writer.drain()
    return Reader.readline().decode()

Now I just need to find out how to properly encode the string I have to send through telnet as so far it triggers an error each time I try to run the script with arguments in dev tools !

Exception in <file.blustream.commutation> line 10: Writer.write('OUT{output}FR{input}\r\n') ^ TypeError: data argument must be a bytes-like object, not 'str'

[craigbarratt]

Yes, the I/O functions expect bytes, not str. You can use the encode() method (default encoding is utf-8):

Writer.write('OUT{output}FR{input}\r\n'.encode())

Like many things in Python, there are several ways to do this (eg: bytes()).

[vincegre]

Yes, the I/O functions expect bytes, not str. You can use the encode() method (default encoding is utf-8):

Thanks a lot for explanations so I modified it to be like that:

@service
def commutation(input=None, output=None):
    log.info("service started")
    import asyncio
    Reader, Writer = None, None
    log.info("telnet starting")
    Reader, Writer = asyncio.open_connection('172.16.2.11', 50005)
    if not Writer:
       raise("Client is not connected")
    Writer.write('OUT{output=}FR{input=}\r\n'.encode())
    Writer.drain()
    return Reader.readline().decode()

Unhappy when I run the script with the 2 arguments it doesn't send the command at device (device doesn't react) and I get an error in logs:

Exception in <file.blustream.commutation> line 12: return Reader.readline().decode() ^ ConnectionResetError: [Errno 104] Connection reset by peer

I got also sometimes that extra error:

Exception in <file.blustream.commutation> line 12: return Reader.readline().decode() ^ UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte

It looks like it's not able to connect at device no ? although it should be able to connect at it without any problems !