Utils¶

Proof State¶

class gym_saturation.proof_state.ProofState(clauses: Dict[str, Dict[str, Any]], step_number: int, max_clauses: int)¶

An object to store all relevant info about a saturation prover state.

Parameters:

clauses – passive clauses
step_number – current step number. -1 before reset, 0 after
max_clauses – maximal possible number of clauses in the proof state

property terminated: bool¶: Refutation found or satisfiability established.

property truncated: bool¶: More clauses generated than expected.

Logic Operations Utility Functions¶

gym_saturation.utils.pretty_print(clause: Dict[str, Any]) → str¶

Print a logical formula back to TPTP language.

Parameters:: clause – a logical clause to print
Returns:: a TPTP string

gym_saturation.utils.start_server_in_a_thread(server: HTTPServer) → None¶

Start a given HTTP server as a daemon.

Parameters:: server – HTTP server

gym_saturation.utils.tptp2python(literals: str) → str¶

Transform a TPTP CNF formula literals to look like Python code.

Parameters:: literals – literals of a TPTP-formatted first-order formula in CNF
Returns:: (hopefully syntactically correct) Python Boolean-valued expression

Vampire Wrapper¶

class gym_saturation.vampire_wrapper.VampireWrapper(binary_path: str, command_line_arguments: str | None = None)¶

A wrapper around Vampire binary running in a manual clause selection mode.

>>> from gym_saturation.constants import (MOCK_TPTP_PROBLEM,
...     MOCK_TPTP_FOLDER)
>>> vampire = VampireWrapper("vampire")
>>> vampire.pick_a_clause("2")
Traceback (most recent call last):
 ...
ValueError: start solving a problem first!
>>> result = vampire.start(MOCK_TPTP_PROBLEM, MOCK_TPTP_FOLDER)

__init__(binary_path: str, command_line_arguments: str | None = None)¶

Remember the path to Vampire, don’t start the process yet.

Parameters:: binary_path – an absolute path to Vampire binary

pick_a_clause(clause_label: str) → Tuple[Tuple[str, str, str], ...]¶

Select a clause and get response from Vampire.

Parameters:: clause_label – a given clause order number
Returns:: a sequence of action type, clause number and clause

property proc: spawn¶

Vampire process.

Raises:: ValueError – when called before reset

start(problem_filename: str, tptp_folder: str) → Tuple[Tuple[str, str, str], ...]¶

Start Vampire in a manual mode on a given problem.

Time limit is one day, Vampire prints everything

Parameters:

problem_filename – full path of a TPTP problem file
tptp_folder – the root folder for TPTP library

Returns:

a sequence of action type, clause number and clause

terminate() → None¶: Terminate Vampire process if any.

Relay Server between Two Sockets¶

class gym_saturation.relay_server.RelayServer(server_address: Tuple[str, int], request_handler_class: Type[BaseRequestHandler], bind_and_activate: bool = True)¶

Server relaying i/o of a long-running connection to a TCP socket to queues.

>>> import socket
>>> from threading import Thread
>>> with RelayServer(("localhost", 0), RelayTCPHandler
... ) as relay_server:
...     thread = Thread(target=relay_server.serve_forever)
...     thread.daemon = True
...     thread.start()
...     with socket.create_connection(relay_server.server_address
...     ) as socket_connection:
...         # data sent to the socket are stored in one queue
...         socket_connection.sendall(bytes(
...             f'{{"tag": "{QUERY_END_MESSAGE}"}}\n\x00\n', "utf8")
...         )
...         print(relay_server.input_queue.get())
...         # data from another queue are sent to the client
...         relay_server.output_queue.put(b"test")
...         print(str(socket_connection.recv(4096), "utf8"))
...         # message format for closing connection
...         socket_connection.sendall(bytes(
...             f'{{"tag": "{SESSION_END_MESSAGE}"}}\n\x00\n', "utf8"
...         ))
...     relay_server.shutdown()
...     thread.join()
{'tag': 'server_queries_start'}
test

__init__(server_address: Tuple[str, int], request_handler_class: Type[BaseRequestHandler], bind_and_activate: bool = True)¶: Initialise queues.

class gym_saturation.relay_server.RelayTCPHandler(request, client_address, server)¶

The request handler class for relay server.

handle() → None¶: Read data from another TCP socket or send data to it.

Dummy HTTP Handler for tests¶

class gym_saturation.dummy_http_handler.DummyHTTPHandler(request, client_address, server)¶

A dummy handler transforming strings to vectors.

do_GET() → None¶: Respond with self._num_features ones as a dummy embedding.

do_POST() → None¶: Respond with self._num_features ones as a dummy embedding.