11426305589

Committed 20 Oct 2024 12:28PM UTC coverage: 82.772%. Remained the same

Build # 11426305589

Build Type

push

github

Committed by

web-flow

Commit Message

Merge 840c77805 into d3660331b

Run Details

1057 of 1277 relevant lines covered (82.77%)

4.9 hits per line

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

92.66

/src/pyswip/prolog.py

# Copyright (c) 2007-2024 Yüce Tekol and PySwip Contributors
#
# 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.

from typing import Union
from pathlib import Path

from pyswip.utils import resolve_path
from pyswip.core import (
    SWI_HOME_DIR,
    PL_STRING,
    REP_UTF8,
    PL_Q_NODEBUG,
    PL_Q_CATCH_EXCEPTION,
    PL_Q_NORMAL,
    PL_initialise,
    PL_open_foreign_frame,
    PL_new_term_ref,
    PL_chars_to_term,
    PL_call,
    PL_discard_foreign_frame,
    PL_new_term_refs,
    PL_put_chars,
    PL_predicate,
    PL_open_query,
    PL_next_solution,
    PL_copy_term_ref,
    PL_exception,
    PL_cut_query,
    PL_thread_self,
    PL_thread_attach_engine,
)


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:
        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) :-
            (read_term_from_atom(GoalString, Goal, [variable_names(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  # noqa: E402


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

yuce / pyswip / 11426305589

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