Running Popper from Python in a loop on different scenarios

[mgualti]

I'm trying to run Popper from Python in a loop like this:

for i in range(len(examples_file_names)):

    popper_settings = popper.util.Settings(
        bias_file = bias_file_name,
        bk_file = background_file_name,
        ex_file = examples_file_names[i])
    prog, score, stats = popper.loop.learn_solution(popper_settings)
    
    prog = list(prog)
    for rule in prog:
        rule_string = popper.util.format_rule(rule)[:-1]
        janus.query_once(f'assert(({rule_string})), fail')

    # Use janus.query to see which examples match the rule...

It appears that some information is not reset between loop iterations. For instance, the examples appear to be accumulating. Is there a way to reset the examples that are stored in the Prolog session, or, better yet, reset everything?

[andrewcropper]

Hi, Yeah, there is an (annoying!) problem with SWI-Prolog. It maintains its state even when it should not. Sorry about that.

Is there a way to reset the examples that are stored in the Prolog session, or, better yet, reset everything?

Not easily. The simplest thing is to create a new Python process each time you create a Popper instance. Kind regards, Andrew

…

On 15 Nov 2024, at 03:13, Marcus Gualtieri ***@***.***> wrote: I'm trying to run Popper from Python in a loop like this: for i in range(len(examples_file_names)): popper_settings = popper.util.Settings( bias_file = bias_file_name, bk_file = background_file_name, ex_file = examples_file_names[i]) prog, score, stats = popper.loop.learn_solution(popper_settings) prog = list(prog) for rule in prog: rule_string = popper.util.format_rule(rule)[:-1] janus.query_once(f'assert(({rule_string})), fail') # Use janus.query to see which examples match the rule... It appears that some information is not reset between loop iterations. For instance, the examples appear to be accumulating. Is there a way to reset the examples that are stored in the Prolog session, or, better yet, reset everything? — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[mgualti]

OK, I will try that. Thanks!

[LedStarr]

Anybody finding themselves in a similar situation this code solved it for me. My main concern was that I intend to run many small popper executions with minor variations and I was afraid I get spillover because of this warning:

Warning: /tmp/popper_j76rq0i2/examples.pl:2:
Warning:    Redefined static procedure pos/1
Warning:    Previously defined at /tmp/popper_a7ib8f6v/examples.pl:2
Warning: /tmp/popper_j76rq0i2/examples.pl:5:
Warning:    Redefined static procedure neg/1
Warning:    Previously defined at /tmp/popper_a7ib8f6v/examples.pl:5
Warning: /tmp/popper_j76rq0i2/bk.pl:2:
Warning:    Redefined static procedure mother/2
Warning:    Previously defined at /tmp/popper_a7ib8f6v/bk.pl:2
Warning: /tmp/popper_j76rq0i2/bk.pl:6:
Warning:    Redefined static procedure father/2
Warning:    Previously defined at /tmp/popper_a7ib8f6v/bk.pl:6

# Popper Demonstration
import asyncio
from dataclasses import dataclass
import os
from os.path import join
from time import sleep
from typing import AsyncGenerator, List, Literal, Tuple
from popper.util import Settings
from popper.loop import learn_solution
import tempfile
from concurrent.futures import ProcessPoolExecutor, as_completed
import janus_swi

MAX_WORKERS = 4


@dataclass
class PopperRequest:
    examples: str
    background_knowledge: str
    bias: str
    temp_dir_prefix: str = "popper_"
    solver: Literal["rc2", "uwr", "wmaxcdcl"] = "rc2"
    verbose = 0


@dataclass
class PopperResult:
    test: str


def validate_input(request: PopperRequest):
    if len(request.examples) < 1:
        return False
    if len(request.background_knowledge) < 1:
        return False
    if len(request.bias) < 1:
        return False
    # TODO: add some actual validation
    return True


def _run_popper(process_index: int, request: PopperRequest):
    # add delay for evaluating async
    if process_index == 0:
        sleep(3)
        
    if not validate_input(request):
        raise ValueError("Given request is not valid.")

    with tempfile.TemporaryDirectory(prefix=request.temp_dir_prefix) as temp_dir:
        with open(os.path.join(temp_dir, "examples.pl"), "w") as f:
            f.write(request.examples)

        with open(os.path.join(temp_dir, "bk.pl"), "w") as f:
            f.write(request.background_knowledge)

        with open(os.path.join(temp_dir, "bias.pl"), "w") as f:
            f.write(request.bias)

        ex_file = join(temp_dir, "examples.pl")
        bk_file = join(temp_dir, "bk.pl")
        bias_file = join(temp_dir, "bias.pl")

        quiet = request.verbose < 1

        settings = Settings(cmd_line=False, quiet=quiet, solver=request.solver, ex_file=ex_file, bk_file=bk_file, bias_file=bias_file)  # pyright: ignore[reportArgumentType]

        prog, score, stats = learn_solution(settings)
        if prog is not None and request.verbose > 0:
            settings.print_prog_score(prog, score)
        if settings.show_stats and request.verbose > 0:
            stats.show()

        # TODO: map results from popper
        result = PopperResult(f"Res: {process_index}")
        return (process_index, result)


# Async wrapper
async def run_popper_async(tasks: List[PopperRequest]) -> AsyncGenerator[Tuple[PopperRequest, PopperResult], None]:
    loop = asyncio.get_running_loop()

    with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
        indexed_futures = [(i, loop.run_in_executor(executor, _run_popper, i, task)) for i, task in enumerate(tasks)]
        futures = [fut for _, fut in indexed_futures]
        for completed in asyncio.as_completed(futures):
            result = await completed
            yield (tasks[result[0]], result[1])


def run_popper(tasks: List[PopperRequest]) -> List[Tuple[PopperRequest, PopperResult]]:
    results = []
    with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(_run_popper, i, task): i for i, task in enumerate(tasks)}
        for future in as_completed(futures):
            i = futures[future]
            res = future.result()
            results.append((tasks[i], res[1]))
    return results

if __name__ == "__main__":
    examples = """
    pos(grandparent(ann,amelia)).
    pos(grandparent(steve,amelia)).
    pos(grandparent(ann,spongebob)).
    neg(grandparent(amy,amelia)).
    """

    bk = """
    mother(ann,amy).
    mother(ann,andy).
    mother(amy,amelia).
    mother(linda,gavin).
    father(steve,amy).
    father(steve,andy).
    father(gavin,amelia).
    father(andy,spongebob).
    """

    bias = """
    head_pred(grandparent,2).
    body_pred(mother,2).
    body_pred(father,2).
    """

    request_1 = PopperRequest(examples, bk, bias)

    examples = """
    pos(grandparent(marie,amelia)).
    pos(grandparent(steve,amelia)).
    pos(grandparent(ann,spongebob)).
    neg(grandparent(amy,amelia)).
    """

    request_2 = PopperRequest(examples, bk, bias)

    # Working
    print("Concurrent execution...")
    for req, result in run_popper([request_1, request_2]):
        print(f"Result for {req}:", result)

    print("Concurrent async execution...")
    async for req, result in run_popper_async([request_1, request_2]):
        print(f"Async result for {req}:", result)

    # Suspect Spillover because of warning
    print("Direct execution...")
    for i, req in enumerate([request_1, request_2]):
        print(_run_popper(i, req))

[rpgoldman]

I would like to this issue because I have a solution that does not rely on multiple Python process spaces, and so does not rely on workers.

Instead, we make the Tester put the program and supporting background knowledge into a temporary module, which can be destroyed when it is no longer needed.

/cc @andrewcropper