From dcdfe0563ab269807f66304107f9123cfc2dd37e Mon Sep 17 00:00:00 2001
From: Timotej Lazar <timotej.lazar@araneo.org>
Date: Thu, 17 Sep 2015 18:01:08 +0200
Subject: Replace prolog.engine.ask_all with check_answers
When testing a program we can stop searching for new solutions after we
have received a wrong one.
---
prolog/engine.py | 43 ++++++++++++++++++++-----------------------
1 file changed, 20 insertions(+), 23 deletions(-)
(limited to 'prolog')
diff --git a/prolog/engine.py b/prolog/engine.py
index 50582c8..eca0826 100644
--- a/prolog/engine.py
+++ b/prolog/engine.py
@@ -6,7 +6,6 @@ import http.client
import json
from operator import itemgetter
import re
-import socket
import time
import urllib
@@ -112,46 +111,45 @@ def pretty_vars(data):
result += [strip_html(b) for b in data['residuals']]
return ',\n'.join(result) if result else 'true'
-# Run [query] in the pengine with id [engine] and return the list of answers
-# found within [timeout] seconds. If a timeout occurs before the query is done,
-# 'timed out' is appended as the last answer.
-def ask_all(engine, query, timeout=10):
- # Returns a tuple ((bindings, constraints), error, more?) for one answer.
+# Run [query] in pengine with ID [engine] and check whether all solutions
+# specified by [answers] (a list of binding dictionaries) are returned. This
+# function succeeds if [query] finds each solution in [answers] at least once
+# within [timeout] seconds, and fails when it finds any other solution.
+def check_answers(engine, query, answers, timeout=10):
+ seen = []
start = time.monotonic()
- answers, messages = [], []
try:
# Run the query.
reply, output = ask(engine, query, timeout)
- messages += output
- if 'error' in map(itemgetter(0), output):
- return answers, messages
answer, error, more = process_answer(reply)
if answer:
- answers.append(answer)
- if error:
- messages.append(error)
+ bindings, constraints = answer
+ if bindings not in answers:
+ return False
+ if bindings not in seen:
+ seen.append(bindings)
# Continue while there are more potential answers and time remaining.
while more:
real_timeout = timeout - (time.monotonic()-start)
if real_timeout <= 0:
- raise socket.timeout()
+ break
reply, output = next(engine, timeout=real_timeout)
- messages += output
answer, error, more = process_answer(reply)
if answer:
- answers.append(answer)
- if error:
- messages.append(error)
- except socket.timeout as ex:
- answers.append('timed out')
- return answers, messages
+ bindings, constraints = answer
+ if bindings not in answers:
+ return False
+ if bindings not in seen:
+ seen.append(bindings)
+ except:
+ pass
+ return len(seen) == len(answers)
# Run [query] in the pengine with id [engine] and return the first answer only
# found within [timeout] seconds.
# used for quick hint debugging by Sasha
def ask_one(engine, query, timeout=1):
- # quicker than ask_all as there could be many hint-triggering tests
# Returns either an error message, true, false, timeout (see below), or bindings
# Timeout is "returned" as an unhandled exception -- deliberately so
@@ -175,7 +173,6 @@ def ask_one(engine, query, timeout=1):
return 'false'
def ask_truth(engine, query, timeout=1):
- # quicker than ask_all as there could be many hint-triggering tests
# Returns either True or False
# (Runtime) error is False!
# Timeout is an unhandled exception -- deliberately so
--
cgit v1.2.1