Detección de presencia por ARP, o cómo saber si tu hijo está en casa (II): Ahora con AsyncIO

#!/usr/bin/env python3

# (c) 2018 Jesús Cea Avión - jcea@jcea.es - https://www.jcea.es/
# This code is licensed under AGPLv3.

import time
import socket
import threading
import asyncio
import random

import telegram.ext


MAC = 'XX:XX:XX:XX:XX:XX'
TELEGRAM_TOKEN = '000000000:XXXXXXXXXXXXXXXXXXXX'
network = '192.168.1.0'
# Usuarios autorizados
# jcea y XXX
usuarios = (000000, 000000)


# Esperamos a tener red porque no queremos que se pierda la
# primera notificación de arranque.
# Además, no queremos un falso negativo cuando aún no tenemos red.
while True:
    try:
        socket.gethostbyname('api.telegram.org')
        break
    except socket.gaierror:
        time.sleep(1)

updater = telegram.ext.updater.Updater(TELEGRAM_TOKEN)
updater.start_polling(timeout=60)
job_queue = updater.job_queue

current_message = ''


def callback(bot, update):
    usuario = update.effective_user['id']
    if usuario in usuarios:
        bot.send_message(usuario, text=current_message, parse_mode='HTML')


updater.dispatcher.add_handler(
        telegram.ext.MessageHandler(
            filters=(telegram.ext.filters.Filters.user(user_id=usuarios) &
                     telegram.ext.filters.Filters.text),
            callback=callback,
            ))


def msg(bot, job):
    texto = job.context['msg']
    for i in usuarios:
        bot.send_message(i, text=texto, parse_mode='HTML')


# Usa TCP
async def conexion2(ip, loop):
    try:
        with socket.socket() as sock:
            sock.setblocking(False)
            fut = loop.sock_connect(sock, (ip, 12345))
            try:
                await asyncio.wait_for(fut, timeout=1)
            except asyncio.TimeoutError:
                pass
    except Exception:
        pass


async def conexion(ip, loop):
    sleep_time = 59.5 * random.random()
    while True:
        await conexion2(ip, loop)
        await asyncio.sleep(sleep_time + random.random(), loop=loop)
        sleep_time = 59.5


async def background_async(first_pass, loop):
    net = '.'.join(network.split('.')[:3])
    futures = []
    for i in range(1, 255):
        ip = net + '.' + str(i)
        futures.append(conexion2(ip, loop))

    await asyncio.wait(futures, loop=loop)
    first_pass.set()

    futures = []
    for i in range(1, 255):
        ip = net + '.' + str(i)
        futures.append(conexion(ip, loop))

    # Esto no termina nunca
    await asyncio.wait(futures, loop=loop)


def background(first_pass):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    task = loop.create_task(background_async(first_pass, loop))
    return loop.run_until_complete(task)


# import subprocess
# def presente():
#     r = subprocess.run(
#             ['/usr/sbin/arp', '-n'],
#             universal_newlines=True,
#             stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
#             timeout=60, check=True)
#     return r.stdout.find(MAC) != -1


# Más eficiente pero no portable
def presente():
    with open('/proc/net/arp') as f:
        tabla_arp = f.read()
    return tabla_arp.find(MAC) != -1


def main():
    global current_message

    presencia = None
    prefix = ' <b>Arranque del sistema:</b>'

    first_pass = threading.Event()
    t = threading.Thread(target=background, args=(first_pass,), daemon=True)
    t.start()

    first_pass.wait()

    while True:
        presencia2 = presente()
        if presencia != presencia2:
            if presencia2:
                current_message = ('%s:%s La luz está encendida'
                                   % (time.ctime(), prefix))
            else:
                current_message = ('%s:%s La luz <b>ESTÁ APAGADA</b>'
                                   % (time.ctime(), prefix))
            print(current_message)

            job_queue.run_once(msg, when=0, context={'msg': current_message})

            presencia = presencia2
            prefix = ''

        time.sleep(60)


if __name__ == '__main__':
    main()