SingularityNET AI Service Integration

class ServiceAdapterABC(ABC):
    """
    This is the service adapter base, all other service adapters are based on it.
    """

    type_name = "Base"

    def __init__(self, app, service: Service, required_services: List[Service] = None) -> None:
        self.app = app
        self.service = service
        self.required_services = required_services
        self.required_service_adapters = []
        self.requirements_met = False
        self.available = False

    def post_load_initialize(self, service_manager: ServiceManager):
        """
        This will hunt out all the agents required to fulfill the required ontology ids

        We should periodically call this if it is false - an agent might come alive that can support this
        :return:
        """
        if not self.required_services is None:
            for required_service in self.required_services:
                service_adapter = service_manager.get_service_adapter_for_id(required_service.node_id)
                self.required_service_adapters.append(service_adapter)
        self.requirements_met = self.has_all_requirements()

        logger.info('Service Adapter: %s initialized. Requirements met: %s', self.type_name, self.requirements_met)
        # print('Service Adapter: %s initialized. Requirements met: %s' % (self.type_name, self.requirements_met))

    def has_all_requirements(self):
        """
        Check to see if our all required services are available
        :return:
        """
        for required_service_adapter in self.required_service_adapters:
            if not required_service_adapter.has_all_requirements():
                return False
        return True

    def start(self):
        """
        If init sets up all the connections, start is here to ensure that the worker is actually alive and can process
        :return:
        """
        self.available = True

    def stop(self):
        """
        This will take the worker offline but does not need to be re-initialized
        :return:
        """
        self.available = False

    def can_perform(self) -> bool:
        """
        This is a boolean flag indicating if this worker can do whatever work it says it can.

        An answer of no can be because it is offline, or perhaps it is too busy.
        :return:
        """
        return self.requirements_met and self.available and self.all_required_agents_can_perform()

    def all_required_agents_can_perform(self):

        if self.required_services is None:
            return True

        for required_service_adapter in self.required_service_adapters:
            if not required_service_adapter.can_perform():
                return False
        return True

    @abstractmethod
    def perform(self, job: JobDescriptor):
        """
        This is where the work gets done, the worker will block here until the work itself is done
        :param args:
        :param kwargs:
        :return:
        """
        pass

class TensorflowMNIST(ServiceAdapterABC):
    type_name = "TensorflowMNIST"

    def __init__(self, app, service: Service, required_services: List[Service] = None):
        super().__init__(app, service, required_services)
        if not service.node_id == MNIST_CLASSIFIER_ID:
            raise RuntimeError("TensorflowMNIST cannot perform service %s", service.node_id)

def post_load_initialize(self, service_manager: ServiceManager):

    # Train the model or load a pretrained model from the cache.

    # There are member variables for the placeholders:
    #     self.input_images - images to be classified
    #     self.keep_prob - controls the dropout during training
    #
    # As well as member variables we need to have during perform:
    #     self.classifier_graph - the tensorflow graph for the image classifier
    #     self.model_path - the location of the cached model

    # Import data
    data_directory = os.path.join(AGENT_DIRECTORY, "input_data")
    mnist_data = input_data.read_data_sets(data_directory, one_hot=True)

    # Create the model - first initialize to default
    tf.reset_default_graph()

    # Define a placeholder for the images
    self.input_images = tf.placeholder(tf.float32, [None, 784])

    # Define loss and optimizer
    input_labels = tf.placeholder(tf.float32, [None, 10])

    # Build the graph for the deep neural network classifier
    self.classifier_graph, self.keep_prob = build_classifier_graph(self.input_images)

    with tf.name_scope('loss'):
        cross_entropy = tf.nn.softmax_cross_entropy_with_logits(labels=input_labels, logits=self.classifier_graph)
    cross_entropy = tf.reduce_mean(cross_entropy)

    with tf.name_scope('adam_optimizer'):
        train_step = tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)

    with tf.name_scope('accuracy'):
        correct_prediction = tf.equal(tf.argmax(self.classifier_graph, 1), tf.argmax(input_labels, 1))
        correct_prediction = tf.cast(correct_prediction, tf.float32)
    accuracy = tf.reduce_mean(correct_prediction)

    # Initialize the variables (i.e. assign their default value)
    initializer = tf.global_variables_initializer()

    # 'Saver' op to save and restore all the variables

    self.model_path = os.path.join(AGENT_DIRECTORY, "model_data", "model.ckpt")
    saver = tf.train.Saver()

    # Create our long-running Tensorflow session
    self.session = tf.Session()

    logger.debug("Checking for pre-trained model in {0}".format(self.model_path))
    if os.path.exists(self.model_path + ".index"):
        logger.debug("Restoring from pre-trained model")

        # Initialize variables
        self.session.run(initializer)

        # Restore model weights from previously saved model
        saver.restore(self.session, self.model_path)

        if CHECK_ACCURACY:
            accuracy = accuracy.eval(feed_dict={
                self.input_images: mnist_data.test.images,
                input_labels: mnist_data.test.labels,
                self.keep_prob: 1.0})
            logger.debug("test accuracy {0}".format(accuracy))

    else:
        logger.debug("No checkpoint - training model from scratch")
        self.session.run(initializer)

        # Train the model
        for i in range(20000):
            batch = mnist_data.train.next_batch(50)
            if i % 100 == 0:
                train_accuracy = accuracy.eval(feed_dict={
                    self.input_images: batch[0], input_labels: batch[1], self.keep_prob: 1.0})
                logger.debug('step {0}, training accuracy {1}'.format(i, train_accuracy))
            train_step.run(feed_dict={
                self.input_images: batch[0],
                input_labels: batch[1],
                self.keep_prob: 0.5})

        # Save model weights to disk
        save_path = saver.save(self.session, self.model_path)
        logger.debug("Model saved in file: {0}".format(save_path))

        if CHECK_ACCURACY:
            accuracy = accuracy.eval(feed_dict={
                self.input_images: mnist_data.test.images,
                input_labels: mnist_data.test.labels,
                self.keep_prob: 1.0})
            logger.debug('Test accuracy {0}'.format(accuracy))

def perform(self, job: JobDescriptor):

        # Process the items in the job. A single job may include a request to classify
        # many different images. Each item, in turn, may be an array of images.
        results = []
        for job_item in job:

            # Make sure the input type is one we can handle...
            input_type = job_item['input_type']
            if input_type != 'attached':
                logger.error("BAD input dict %s", str(job_item))
                raise RuntimeError("TensorflowMNIST - job item 'input_type' must be 'attached'.")

            # Get the images to classify, while making sure our job item dict is of the appropriate format.
            input_data = job_item['input_data']
            if input_data is None:
                raise RuntimeError("TensorflowMNIST - job item 'input_data' must be defined.")
            images_to_classify = input_data.get('images')
            if images_to_classify is None:
                raise RuntimeError("TensorflowMNIST - job item 'input_data' missing 'images'")

            # Get the predication and confidence for each image in this job item
            prediction = tf.argmax(self.classifier_graph, 1)
            confidence = tf.nn.softmax(self.classifier_graph)
            predictions = prediction.eval(session=self.session,
                                          feed_dict={self.input_images: images_to_classify, self.keep_prob: 1.0})
            confidences = confidence.eval(session=self.session,
                                          feed_dict={self.input_images: images_to_classify, self.keep_prob: 1.0})
            prediction_confidences = []
            for index in range(0, len(images_to_classify)):
                prediction_confidence = confidences[index][predictions[index]]
                prediction_confidences.append(float(prediction_confidence))

            logger.debug("Predictions: {0}".format(predictions))
            logger.debug("Confidences: {0}".format(prediction_confidences))

            # Add the job results to our combined results array for all job items.
            single_job_result = {
                'predictions': predictions.tolist(),
                'confidences': prediction_confidences,
            }
            results.append(single_job_result)

        return results