yuce / pyswip / 110

pending completion

Build # 110

Build Type

Pull #97

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/yuce/pyswip/commit/<a class=hub.com/yuce/pyswip/commit/<a class="double-link" href="https://git"><a class=hub.com/yuce/pyswip/commit/<a class="double-link" href="https://git"><a class=hub.com/yuce/pyswip/commit/<a class="double-link" href="https://git"><a class=hub.com/yuce/pyswip/commit/db4456c740fbb1a8ed14982bff5869c93551e8ad">db4456c74">&lt;a href=&quot;https://github.com/yuce/pyswip/commit/</a><a class="double-link" href="https://github.com/yuce/pyswip/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/yuce/pyswip/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/yuce/pyswip/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/yuce/pyswip/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/yuce/pyswip/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/yuce/pyswip/commit/db4456c740fbb1a8ed14982bff5869c93551e8ad">db4456c74</a><a href="https://github.com/yuce/pyswip/commit/db4456c740fbb1a8ed14982bff5869c93551e8ad">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/yuce/pyswip/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/yuce/pyswip/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/yuce/pyswip/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/yuce/pyswip/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/yuce/pyswip/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/yuce/pyswip/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/yuce/pyswip/commit/db4456c740fbb1a8ed14982bff5869c93551e8ad&quot;&gt;db4456c74&lt;/a&gt;&lt;a href=&quot;https://github.com/yuce/pyswip/commit/db4456c740fbb1a8ed14982bff5869c93551e8ad&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/yuce/pyswip/co... (continued)

Pull Request Pull Request #97: Improve list representations, unicode support and multiple threading usage

Run Details

79 of 79 new or added lines in 3 files covered. (100.0%)

906 of 1150 relevant lines covered (78.78%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.52

/pyswip/prolog.py

# -*- coding: utf-8 -*-


# prolog.py -- Prolog class
# Copyright (c) 2007-2012 Yüce Tekol
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


import sys

from pyswip.core import *


class PrologError(Exception):
    pass


class NestedQueryError(PrologError):
    """
    SWI-Prolog does not accept nested queries, that is, opening a query while
    the previous one was not closed.

    As this error may be somewhat difficult to debug in foreign code, it is
    automatically treated inside pySWIP
    """
    pass


def _initialize():
    args = []
    args.append("./")
    args.append("-q")         # --quiet
    args.append("--nosignals") # "Inhibit any signal handling by Prolog"
    if SWI_HOME_DIR is not None:
        args.append("--home=%s" % SWI_HOME_DIR)

    result = PL_initialise(len(args),args)
    # result is a boolean variable (i.e. 0 or 1) indicating whether the
    # initialisation was successful or not.
    if not result:
        raise PrologError("Could not initialize the Prolog environment."
                          "PL_initialise returned %d" % result)

    swipl_fid = PL_open_foreign_frame()
    swipl_load = PL_new_term_ref()
    PL_chars_to_term("asserta(pyrun(GoalString,BindingList) :- "
                     "(atom_chars(A,GoalString),"
                     "atom_to_term(A,Goal,BindingList),"
                     "call(Goal))).", swipl_load)
    PL_call(swipl_load, None)
    PL_discard_foreign_frame(swipl_fid)
_initialize()


# NOTE: This import MUST be after _initialize is called!!
from pyswip.easy import getTerm


class Prolog:
    """Easily query SWI-Prolog.
    This is a singleton class
    """

    # We keep track of open queries to avoid nested queries.
    _queryIsOpen = False

    class _QueryWrapper(object):

        def __init__(self):
            if Prolog._queryIsOpen:
                raise NestedQueryError("The last query was not closed")

        def __call__(self, query, maxresult, catcherrors, normalize):
            Prolog._init_prolog_thread()
            swipl_fid = PL_open_foreign_frame()

            swipl_head = PL_new_term_ref()
            swipl_args = PL_new_term_refs(2)
            swipl_goalCharList = swipl_args
            swipl_bindingList = swipl_args + 1

            PL_put_chars(swipl_goalCharList, PL_STRING|REP_UTF8, -1, query.encode("utf-8"))

            swipl_predicate = PL_predicate("pyrun", 2, None)

            plq = catcherrors and (PL_Q_NODEBUG|PL_Q_CATCH_EXCEPTION) or PL_Q_NORMAL
            swipl_qid = PL_open_query(None, plq, swipl_predicate, swipl_args)

            Prolog._queryIsOpen = True # From now on, the query will be considered open
            try:
                while maxresult and PL_next_solution(swipl_qid):
                    maxresult -= 1
                    bindings = []
                    swipl_list = PL_copy_term_ref(swipl_bindingList)
                    t = getTerm(swipl_list)
                    if normalize:
                        try:
                            v = t.value
                        except AttributeError:
                            v = {}
                            for r in [x.value for x in t]:
                                r = normalize_values(r)
                                v.update(r)
                        yield v
                    else:
                        yield t

                if PL_exception(swipl_qid):
                    term = getTerm(PL_exception(swipl_qid))

                    raise PrologError("".join(["Caused by: '", query, "'. ",
                                               "Returned: '", str(term), "'."]))

            finally: # This ensures that, whatever happens, we close the query
                PL_cut_query(swipl_qid)
                PL_discard_foreign_frame(swipl_fid)
                Prolog._queryIsOpen = False

    @classmethod
    def _init_prolog_thread(cls):
        pengine_id = PL_thread_self()
        if pengine_id == -1:
            pengine_id = PL_thread_attach_engine(None)
        if pengine_id == -1:
            raise PrologError("Unable to attach new Prolog engine to the thread")
        elif pengine_id == -2:
            print("{WARN} Single-threaded swipl build, beware!")

    @classmethod
    def asserta(cls, assertion, catcherrors=False):
        next(cls.query(assertion.join(["asserta((", "))."]), catcherrors=catcherrors))

    @classmethod
    def assertz(cls, assertion, catcherrors=False):
        next(cls.query(assertion.join(["assertz((", "))."]), catcherrors=catcherrors))

    @classmethod
    def dynamic(cls, term, catcherrors=False):
        next(cls.query(term.join(["dynamic((", "))."]), catcherrors=catcherrors))

    @classmethod
    def retract(cls, term, catcherrors=False):
        next(cls.query(term.join(["retract((", "))."]), catcherrors=catcherrors))

    @classmethod
    def retractall(cls, term, catcherrors=False):
        next(cls.query(term.join(["retractall((", "))."]), catcherrors=catcherrors))

    @classmethod
    def consult(cls, filename, catcherrors=False):
        next(cls.query(filename.join(["consult('", "')"]), catcherrors=catcherrors))

    @classmethod
    def query(cls, query, maxresult=-1, catcherrors=True, normalize=True):
        """Run a prolog query and return a generator.
        If the query is a yes/no question, returns {} for yes, and nothing for no.
        Otherwise returns a generator of dicts with variables as keys.

        >>> prolog = Prolog()
        >>> prolog.assertz("father(michael,john)")
        >>> prolog.assertz("father(michael,gina)")
        >>> bool(list(prolog.query("father(michael,john)")))
        True
        >>> bool(list(prolog.query("father(michael,olivia)")))
        False
        >>> print sorted(prolog.query("father(michael,X)"))
        [{'X': 'gina'}, {'X': 'john'}]
        """
        return cls._QueryWrapper()(query, maxresult, catcherrors, normalize)


def normalize_values(values):
    from pyswip.easy import Atom, Functor
    if isinstance(values, Atom):
        return values.value
    if isinstance(values, Functor):
        normalized = values.name.value
        if values.arity:
            normalized_args = ([str(normalize_values(arg)) for arg in values.args])
            normalized = normalized + '(' + ', '.join(normalized_args) + ')'
        return normalized
    elif isinstance(values, dict):
        return {key: normalize_values(v) for key, v in values.items()}
    elif isinstance(values, (list, tuple)):
        return [normalize_values(v) for v in values]
    return values


1	# -- coding: utf-8 --
2
3
4	# prolog.py -- Prolog class
5	# Copyright (c) 2007-2012 Yüce Tekol
6	#
7	# Permission is hereby granted, free of charge, to any person obtaining a copy
8	# of this software and associated documentation files (the "Software"), to deal
9	# in the Software without restriction, including without limitation the rights
10	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11	# copies of the Software, and to permit persons to whom the Software is
12	# furnished to do so, subject to the following conditions:
13	#
14	# The above copyright notice and this permission notice shall be included in all
15	# copies or substantial portions of the Software.
16	#
17	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23	# SOFTWARE.
24
25
26	import sys	1✔
27
28	from pyswip.core import *	1✔
29
30
31	class PrologError(Exception):	1✔
32	pass	1✔
33
34
35	class NestedQueryError(PrologError):	1✔
36	"""
37	SWI-Prolog does not accept nested queries, that is, opening a query while
38	the previous one was not closed.
39
40	As this error may be somewhat difficult to debug in foreign code, it is
41	automatically treated inside pySWIP
42	"""
43	pass	1✔
44
45
46	def _initialize():	1✔
47	args = []	1✔
48	args.append("./")	1✔
49	args.append("-q") # --quiet	1✔
50	args.append("--nosignals") # "Inhibit any signal handling by Prolog"	1✔
51	if SWI_HOME_DIR is not None:	1✔
52	args.append("--home=%s" % SWI_HOME_DIR)	1✔
53
54	result = PL_initialise(len(args),args)	1✔
55	# result is a boolean variable (i.e. 0 or 1) indicating whether the
56	# initialisation was successful or not.
57	if not result:	1✔
58	raise PrologError("Could not initialize the Prolog environment."	×
59	"PL_initialise returned %d" % result)
60
61	swipl_fid = PL_open_foreign_frame()	1✔
62	swipl_load = PL_new_term_ref()	1✔
63	PL_chars_to_term("asserta(pyrun(GoalString,BindingList) :- "	1✔
64	"(atom_chars(A,GoalString),"
65	"atom_to_term(A,Goal,BindingList),"
66	"call(Goal))).", swipl_load)
67	PL_call(swipl_load, None)	1✔
68	PL_discard_foreign_frame(swipl_fid)	1✔
69	_initialize()	1✔
70
71
72	# NOTE: This import MUST be after _initialize is called!!
73	from pyswip.easy import getTerm	1✔
74
75
76	class Prolog:	1✔
77	"""Easily query SWI-Prolog.
78	This is a singleton class
79	"""
80
81	# We keep track of open queries to avoid nested queries.
82	_queryIsOpen = False	1✔
83
84	class _QueryWrapper(object):	1✔
85
86	def __init__(self):	1✔
87	if Prolog._queryIsOpen:	1✔
88	raise NestedQueryError("The last query was not closed")	1✔
89
90	def __call__(self, query, maxresult, catcherrors, normalize):	1✔
91	Prolog._init_prolog_thread()	1✔
92	swipl_fid = PL_open_foreign_frame()	1✔
93
94	swipl_head = PL_new_term_ref()	1✔
95	swipl_args = PL_new_term_refs(2)	1✔
96	swipl_goalCharList = swipl_args	1✔
97	swipl_bindingList = swipl_args + 1	1✔
98
99	PL_put_chars(swipl_goalCharList, PL_STRING\|REP_UTF8, -1, query.encode("utf-8"))	1✔
100
101	swipl_predicate = PL_predicate("pyrun", 2, None)	1✔
102
103	plq = catcherrors and (PL_Q_NODEBUG\|PL_Q_CATCH_EXCEPTION) or PL_Q_NORMAL	1✔
104	swipl_qid = PL_open_query(None, plq, swipl_predicate, swipl_args)	1✔
105
106	Prolog._queryIsOpen = True # From now on, the query will be considered open	1✔
107	try:	1✔
108	while maxresult and PL_next_solution(swipl_qid):	1✔
109	maxresult -= 1	1✔
110	bindings = []	1✔
111	swipl_list = PL_copy_term_ref(swipl_bindingList)	1✔
112	t = getTerm(swipl_list)	1✔
113	if normalize:	1✔
114	try:	1✔
115	v = t.value	1✔
116	except AttributeError:	1✔
117	v = {}	1✔
118	for r in [x.value for x in t]:	1✔
119	r = normalize_values(r)	1✔
120	v.update(r)	1✔
121	yield v	1✔
122	else:
123	yield t	×
124
125	if PL_exception(swipl_qid):	1✔
126	term = getTerm(PL_exception(swipl_qid))	×
127
128	raise PrologError("".join(["Caused by: '", query, "'. ",	×
129	"Returned: '", str(term), "'."]))
130
131	finally: # This ensures that, whatever happens, we close the query
132	PL_cut_query(swipl_qid)	1✔
133	PL_discard_foreign_frame(swipl_fid)	1✔
134	Prolog._queryIsOpen = False	1✔
135
136	@classmethod	1✔
137	def _init_prolog_thread(cls):
138	pengine_id = PL_thread_self()	1✔
139	if pengine_id == -1:	1✔
140	pengine_id = PL_thread_attach_engine(None)	×
141	if pengine_id == -1:	1✔
142	raise PrologError("Unable to attach new Prolog engine to the thread")	×
143	elif pengine_id == -2:	1✔
144	print("{WARN} Single-threaded swipl build, beware!")	×
145
146	@classmethod	1✔
147	def asserta(cls, assertion, catcherrors=False):	1✔
148	next(cls.query(assertion.join(["asserta((", "))."]), catcherrors=catcherrors))	×
149
150	@classmethod	1✔
151	def assertz(cls, assertion, catcherrors=False):	1✔
152	next(cls.query(assertion.join(["assertz((", "))."]), catcherrors=catcherrors))	1✔
153
154	@classmethod	1✔
155	def dynamic(cls, term, catcherrors=False):	1✔
156	next(cls.query(term.join(["dynamic((", "))."]), catcherrors=catcherrors))	1✔
157
158	@classmethod	1✔
159	def retract(cls, term, catcherrors=False):	1✔
160	next(cls.query(term.join(["retract((", "))."]), catcherrors=catcherrors))	1✔
161
162	@classmethod	1✔
163	def retractall(cls, term, catcherrors=False):	1✔
164	next(cls.query(term.join(["retractall((", "))."]), catcherrors=catcherrors))	1✔
165
166	@classmethod	1✔
167	def consult(cls, filename, catcherrors=False):	1✔
168	next(cls.query(filename.join(["consult('", "')"]), catcherrors=catcherrors))	1✔
169
170	@classmethod	1✔
171	def query(cls, query, maxresult=-1, catcherrors=True, normalize=True):	1✔
172	"""Run a prolog query and return a generator.
173	If the query is a yes/no question, returns {} for yes, and nothing for no.
174	Otherwise returns a generator of dicts with variables as keys.
175
176	>>> prolog = Prolog()
177	>>> prolog.assertz("father(michael,john)")
178	>>> prolog.assertz("father(michael,gina)")
179	>>> bool(list(prolog.query("father(michael,john)")))
180	True
181	>>> bool(list(prolog.query("father(michael,olivia)")))
182	False
183	>>> print sorted(prolog.query("father(michael,X)"))
184	[{'X': 'gina'}, {'X': 'john'}]
185	"""
186	return cls._QueryWrapper()(query, maxresult, catcherrors, normalize)	1✔
187
188
189	def normalize_values(values):	1✔
190	from pyswip.easy import Atom, Functor	1✔
191	if isinstance(values, Atom):	1✔
192	return values.value	1✔
193	if isinstance(values, Functor):	1✔
194	normalized = values.name.value	1✔
195	if values.arity:	1✔
196	normalized_args = ([str(normalize_values(arg)) for arg in values.args])	1✔
197	normalized = normalized + '(' + ', '.join(normalized_args) + ')'	1✔
198	return normalized	1✔
199	elif isinstance(values, dict):	1✔
200	return {key: normalize_values(v) for key, v in values.items()}	1✔
201	elif isinstance(values, (list, tuple)):	1✔
202	return [normalize_values(v) for v in values]	1✔
203	return values	1✔
204

yuce / pyswip / 110

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous