aspartix_interface.py

"""Copyright 2017 Ziyi Bao, Department of Computing, Imperial College London

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License."""

"""
This module contains a class ASPARTIX_Interface that acts as an interface between an ABA_Plus object and
ASP solvers, thus enabling the computation of extensions under various semantics.
"""

import subprocess
import re
import sys
import os

from sys import platform as _platform

MODULE_DIR = os.path.dirname(sys.modules[__name__].__file__)
DLV = "dlv"
# We will join DLV with the MODULE_DIR in calculate_extensions().
DLV_IDEAL_COMMAND = DLV + " {} {} -filter=ideal -n=1"

CLINGO = "clingo"

CLINGO_COMMAND = "clingo {} {} 0"


CLINGO_ANSWER = "Answer:"
DLV_ANSWER = "Best model:"

CLINGO_REGEX = r"in\((\d+)\)"
DLV_IDEAL_REGEX = r"ideal\((\d+)\)"

ADMISSIBLE_FILE = "adm.dl"
STABLE_FILE= "stable.dl"
IDEAL_FILE = "ideal.dl"
COMPLETE_FILE = "comp.dl"
PREFERRED_FILE = "prefex_gringo.lp"
GROUNDED_FILE = "ground.dl"



class ASPARTIX_Interface:
    def __init__(self, aba_plus):
        """
        :param aba_plus: ABA_Plus object for which the calculation of extensions are performed
        """
        self.aba_plus = aba_plus

    def generate_input_file_for_clingo(self, filename):
        """
        generate from the ABA+ framework (self.aba_plus) an input file that can be fed into an ASP solver
        :param filename: save generated file under filename
        """
        res = self.aba_plus.generate_arguments_and_attacks_for_contraries()
        deductions = res[2]
        attacks = res[1]

        #maps arguments to indices, which are used to represent the arguments in the input file
        self.arguments = []
        for deduction in deductions:
            if deduction.premise not in self.arguments:
                self.arguments.append(deduction.premise)

        self.attacks = set()
        for atk in attacks:
            self.attacks.add((frozenset(atk.attacker.premise),
                              frozenset(atk.attackee.premise)))

        f = open(filename, 'w')

        for idx in range(0, len(self.arguments)):
            f.write("arg({}).\n".format(idx))

        for atk in self.attacks:
            idx_attacker = self.arguments.index(atk[0])
            idx_attackee = self.arguments.index(atk[1])
            f.write("att({}, {}).\n".format(idx_attacker, idx_attackee))

        f.close()

    def calculate_admissible_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of admissible sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(CLINGO_COMMAND, input_filename, ADMISSIBLE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_stable_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of stable sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(CLINGO_COMMAND, input_filename, STABLE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_ideal_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of ideal sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(DLV_IDEAL_COMMAND, input_filename, IDEAL_FILE, DLV_ANSWER, DLV_IDEAL_REGEX)

    def calculate_complete_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of complete sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(CLINGO_COMMAND, input_filename, COMPLETE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_preferred_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of preferred sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(CLINGO_COMMAND, input_filename, PREFERRED_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_grounded_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: the set of grounded sets of Sentences(assumptions) under the ABA+ framework (self.aba_plus)
        """
        return self.calculate_extensions(CLINGO_COMMAND, input_filename, GROUNDED_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    subprocess_has_run = hasattr(subprocess, 'run')
    def calculate_extensions(self, command, input_filename, encoding_filename, answer_header, regex):
        """
        :param command: command to run the desired ASP solver. If the command is
          DLV, the executable should be in this MODULE_DIR.
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into the desired ASP solver
        :param encoding_filename: name of the file that encodes the desired semantics, which should
          not contain spaces, and which should be in this MODULE_DIR.
        :param answer_header: answer head that the desired solver outputs
        :param regex: regular expression matching the answer symbols
        :return: the set of sets of Sentences(assumptions) under the semantics encoded by encoding_filename
        """

        args = command.format(input_filename, encoding_filename).split(" ")
        args[args.index(encoding_filename)] = os.path.join(MODULE_DIR, encoding_filename)
        if DLV in args:
            args[args.index(DLV)] = os.path.join(MODULE_DIR, DLV)

        if ASPARTIX_Interface.subprocess_has_run:
            #for python 3.5 and later:
            res = subprocess.run(args,
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                 universal_newlines=True)

            if answer_header not in res.stdout:
                return set()


            results = res.stdout.split(answer_header)
        else:
            res = subprocess.Popen(args,
                                   stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            output = res.stdout.read().decode("utf-8")

            if answer_header not in output:
                return set()


            results = output.split(answer_header)

        extension_sets = set()
        answer_sets = results[1:len(results)+1]
        for answer in answer_sets:
            matches = re.findall(regex, answer)
            extension = set()
            for m in matches:
                arg = self.arguments[int(m)]
                extension = extension.union(arg)
            extension_sets.add(frozenset(extension))

        return extension_sets

    def calculate_admissible_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping admissible sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(CLINGO_COMMAND, input_filename, ADMISSIBLE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_stable_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping stable sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(CLINGO_COMMAND, input_filename, STABLE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_ideal_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping ideal sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(DLV_IDEAL_COMMAND, input_filename, IDEAL_FILE, DLV_ANSWER, DLV_IDEAL_REGEX)

    def calculate_complete_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping complete sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(CLINGO_COMMAND, input_filename, COMPLETE_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_preferred_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping preferred sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(CLINGO_COMMAND, input_filename, PREFERRED_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_grounded_arguments_extensions(self, input_filename):
        """
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into an ASP solver
        :return: dictionary mapping grounded sets under self.abap_plus to their conclusions
        """
        return self.calculate_arguments_extensions(CLINGO_COMMAND, input_filename, GROUNDED_FILE, CLINGO_ANSWER, CLINGO_REGEX)

    def calculate_arguments_extensions(self, command, input_filename, encoding_filename, answer_header, regex):
        """
        :param command: command to run the desired ASP solver. If the command is
          DLV, the executable should be in this MODULE_DIR.
        :param input_filename: name of the file generated by generate_input_file_for_clingo(),
                               the file will be fed into the desired ASP solver
        :param encoding_filename: name of the file that encodes the desired semantics, which should
          not contain spaces, and which should be in this MODULE_DIR.
        :param answer_header: answer head that the desired solver outputs
        :param regex: regular expression matching the answer symbols
        :return: dictionary mapping sets under the semantics encoded by encoding_filename to their conclusions

        """

        args = command.format(input_filename, encoding_filename).split(" ")
        args[args.index(encoding_filename)] = os.path.join(MODULE_DIR, encoding_filename)
        if DLV in args:
            args[args.index(DLV)] = os.path.join(MODULE_DIR, DLV)

        res = subprocess.Popen(args,
                               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        output = res.stdout.read().decode("utf-8")

        if answer_header not in output:
            return {}

        results = output.split(answer_header)

        # maps sets of sentences to sets of conclusions
        extension_dict = {}
        answer_sets = results[1:len(results)+1]
        for answer in answer_sets:
            matches = re.findall(regex, answer)
            extension = set()
            conclusions = set()
            for m in matches:
                arg = self.arguments[int(m)]
                extension = extension.union(arg)
            conclusions = self.aba_plus.generate_all_deductions(extension)
            extension = frozenset(extension)
            if extension in extension_dict:
                extension_dict[extension] =  extension_dict[extension].union(conclusions)
            else:
                extension_dict[extension] = conclusions
        return extension_dict