Source code for festivalgrid.management.commands.update_devices

import time
import json
import socket
import urllib
import urllib.error
import multiprocessing.dummy
import urllib.request
import ipaddress

from django.core.management.base import BaseCommand

from ...models import Device


class Command(BaseCommand):
    help = 'Scan for devices'

    def add_arguments(self, parser):
        parser.add_argument('ip4range', type=str)
        parser.add_argument('--timeout', type=int, nargs='?', const=5)

    def handle(self, *args, **options):
        self.stdout.write("\nScanning %s...\n" % options["ip4range"])

        def runner(ip):
            try:
                url = 'http://' + str(ip) + '/cm?cmnd=status%200'
                with urllib.request.urlopen(url, timeout=options["timeout"]) as request:
                    status = json.loads(request.read())
                    self.stdout.write(str(status))
                    device, created = Device.objects.update_or_create(
                        mac=status["StatusNET"]["Mac"], defaults={
                            "template": "tasmota",
                            "name": status["Status"]["FriendlyName"][0],
                            "ip": str(ip),
                            "hostname": status["StatusNET"]["Hostname"],
                            "mqtt_topic": status["Status"]["Topic"]
                        }
                    )
                    self.stdout.write(
                        self.style.SUCCESS("\n%s: %s\n" % (ip, status["Status"]["FriendlyName"][0])) +
                        json.dumps(status, sort_keys=True, indent=4) + "\n" +
                        self.style.MIGRATE_LABEL("%s device." % ("Created new" if created else "Updated existing")) + "\n\n"
                    )
            except urllib.error.URLError:
                time.sleep(1)
                self.stdout.write(self.style.ERROR("%s: not a tasmota device" % ip))
            except (TimeoutError, socket.timeout):
                self.stdout.write(self.style.ERROR("%s: timed out" % ip))

        p = multiprocessing.dummy.Pool(255)
        p.map(runner, ipaddress.IPv4Network(options["ip4range"]))

        self.stdout.write("\nscan completed.")