OpenDroneMap-WebODM/nodeodm/models.py

from __future__ import unicode_literals

from django.db import models
from django.contrib.postgres import fields
from django.utils import timezone
from django.dispatch import receiver
from guardian.models import GroupObjectPermissionBase
from guardian.models import UserObjectPermissionBase
from django.utils.translation import gettext_lazy as _

from webodm import settings

import json
from pyodm import Node
from pyodm import exceptions
from django.db.models import signals
from datetime import timedelta
import logging

logger = logging.getLogger('app.logger')

class ProcessingNode(models.Model):
    hostname = models.CharField(verbose_name=_("Hostname"), max_length=255, help_text=_("Hostname or IP address where the node is located (can be an internal hostname as well). If you are using Docker, this is never 127.0.0.1 or localhost. Find the IP address of your host machine by running ifconfig on Linux or by checking your network settings."))
    port = models.PositiveIntegerField(verbose_name=_("Port"), help_text=_("Port that connects to the node's API"))
    api_version = models.CharField(verbose_name=_("API Version"), max_length=32, null=True, help_text=_("API version used by the node"))
    last_refreshed = models.DateTimeField(verbose_name=_("Last Refreshed"), null=True, help_text=_("When was the information about this node last retrieved?"))
    queue_count = models.PositiveIntegerField(verbose_name=_("Queue Count"), default=0, help_text=_("Number of tasks currently being processed by this node (as reported by the node itself)"))
    available_options = fields.JSONField(verbose_name=_("Available Options"), default=dict, help_text=_("Description of the options that can be used for processing"))
    token = models.CharField(verbose_name=_("Token"), max_length=1024, blank=True, default="", help_text=_("Token to use for authentication. If the node doesn't have authentication, you can leave this field blank."))
    max_images = models.PositiveIntegerField(verbose_name=_("Max Images"), help_text=_("Maximum number of images accepted by this node."), blank=True, null=True)
    engine_version = models.CharField(verbose_name=_("Engine Version"), max_length=32, null=True, help_text=_("Engine version used by the node."))
    label = models.CharField(verbose_name=_("Label"), max_length=255, default="", blank=True, help_text=_("Optional label for this node. When set, this label will be shown instead of the hostname:port name."))
    engine = models.CharField(verbose_name=_("Engine"), max_length=255, null=True, help_text=_("Engine used by the node."))

    class Meta:
        verbose_name = _("Processing Node")
        verbose_name_plural = _("Processing Nodes")

    def __str__(self):
        if self.label != "":
            return self.label
        else:
            return '{}:{}'.format(self.hostname, self.port)

    @staticmethod
    def find_best_available_node():
        """
        Attempts to find an available node (seen in the last 5 minutes, and with lowest queue count)
        :return: ProcessingNode | None
        """
        return ProcessingNode.objects.filter(last_refreshed__gte=timezone.now() - timedelta(minutes=settings.NODE_OFFLINE_MINUTES)) \
                                     .order_by('queue_count').first()

    def is_online(self):
        if settings.NODE_OPTIMISTIC_MODE:
            return True

        return self.last_refreshed is not None and \
               self.last_refreshed >= timezone.now() - timedelta(minutes=settings.NODE_OFFLINE_MINUTES)

    def update_node_info(self):
        """
        Retrieves information and options from the node API
        and saves it into the database.

        :returns: True if information could be updated, False otherwise
        """
        api_client = self.api_client(timeout=5)
        try:
            info = api_client.info()

            self.api_version = info.version
            self.queue_count = info.task_queue_count
            self.max_images = info.max_images
            self.engine_version = info.engine_version
            self.engine = info.engine

            options = list(map(lambda o: o.__dict__, api_client.options()))
            self.available_options = options
            self.last_refreshed = timezone.now()
            self.save()
            return True
        except exceptions.OdmError:
            return False

    def api_client(self, timeout=30):
        return Node(self.hostname, self.port, self.token, timeout)

    def get_available_options_json(self, pretty=False):
        """
        :returns available options in JSON string format
        """
        kwargs = dict(indent=4, separators=(',', ": ")) if pretty else dict() 
        return json.dumps(self.available_options, **kwargs)

    def options_list_to_dict(self, options = []):
        """
        Convers options formatted as a list ([{'name': optionName, 'value': optionValue}, ...])
        to a dictionary {optionName: optionValue, ...}
        :param options: options
        :return: dict
        """
        opts = {}
        if options is not None:
            for o in options:
                opts[o['name']] = o['value']

        return opts

    def process_new_task(self, images, name=None, options=[], progress_callback=None):
        """
        Sends a set of images (and optional GCP file) via the API
        to start processing.

        :param images: list of path images
        :param name: name of the task
        :param options: options to be used for processing ([{'name': optionName, 'value': optionValue}, ...])
        :param progress_callback: optional callback invoked during the upload images process to be used to report status.

        :returns UUID of the newly created task
        """
        if len(images) < 1: raise exceptions.NodeServerError("Need at least 1 file")

        api_client = self.api_client()

        opts = self.options_list_to_dict(options)

        task = api_client.create_task(images, opts, name, progress_callback)
        return task.uuid

    def get_task_info(self, uuid, with_output=None):
        """
        Gets information about this task, such as name, creation date, 
        processing time, status, command line options and number of 
        images being processed.
        """
        api_client = self.api_client()
        task = api_client.get_task(uuid)
        task_info = task.info(with_output)

        # Output support for older clients
        if not api_client.version_greater_or_equal_than("1.5.1") and with_output:
            task_info.output = self.get_task_console_output(uuid, with_output)

        return task_info

    def get_task_console_output(self, uuid, line):
        """
        Retrieves the console output of the OpenDroneMap's process.
        Useful for monitoring execution and to provide updates to the user.
        """
        api_client = self.api_client()
        task = api_client.get_task(uuid)
        return task.output(line)

    def cancel_task(self, uuid):
        """
        Cancels a task (stops its execution, or prevents it from being executed)
        """
        api_client = self.api_client()
        task = api_client.get_task(uuid)
        return task.cancel()

    def remove_task(self, uuid):
        """
        Removes a task and deletes all of its assets
        """
        api_client = self.api_client()
        task = api_client.get_task(uuid)
        return task.remove()

    def download_task_assets(self, uuid, destination, progress_callback, parallel_downloads=16):
        """
        Downloads a task asset
        """
        api_client = self.api_client()
        task = api_client.get_task(uuid)
        return task.download_zip(destination, progress_callback, parallel_downloads=parallel_downloads)

    def restart_task(self, uuid, options = None):
        """
        Restarts a task that was previously canceled or that had failed to process
        """

        api_client = self.api_client()
        task = api_client.get_task(uuid)
        return task.restart(self.options_list_to_dict(options))

    def delete(self, using=None, keep_parents=False):
        pnode_id = self.id
        super(ProcessingNode, self).delete(using, keep_parents)

        from app.plugins import signals as plugin_signals
        plugin_signals.processing_node_removed.send_robust(sender=self.__class__, processing_node_id=pnode_id)


# First time a processing node is created, automatically try to update
@receiver(signals.post_save, sender=ProcessingNode, dispatch_uid="update_processing_node_info")
def auto_update_node_info(sender, instance, created, **kwargs):
    if created:
        try:
            instance.update_node_info()
        except exceptions.OdmError:
            pass
        except Exception as e:
            logger.warning("auto_update_node_info: " + str(e))

class ProcessingNodeUserObjectPermission(UserObjectPermissionBase):
    content_object = models.ForeignKey(ProcessingNode, on_delete=models.CASCADE)


class ProcessingNodeGroupObjectPermission(GroupObjectPermissionBase):
    content_object = models.ForeignKey(ProcessingNode, on_delete=models.CASCADE)