LiSE version 0.16

clayot

The 0.15 series of LiSE releases was all about dynamic unloading, which wasn't much fun to show off because it just meant LiSE would use less memory. And time travel was slower in some cases.

0.16 is about multiprocess parallelism, which turned out to be easier to implement than I expected. Here's how you find a bunch of paths all at once in a LiSE sim:

import random

import networkx as nx

from LiSE import Engine


with Engine(random_seed=69105) as eng:
    random.seed(69105)
    grid: nx.Graph = nx.grid_2d_graph(100, 100)

    for node in list(grid):
        if random.random() < 0.1:
            grid.remove_node(node)
        elif random.random() < 0.01:
            grid.add_node(f"{node}_inhabitant", location=node)

    phys = eng.new_character("physical", grid)

    @eng.function
    def find_path_somewhere(node):
        from networkx.algorithms import astar_path
        from math import sqrt

        x, y = node.location.name
        destx = 100 - int(x)
        desty = 100 - int(y)
        while (destx, desty) not in node.character.place:
            if destx < 99:
                destx += 1
            elif desty < 99:
                destx = 0
                desty += 1
            else:
                destx = desty = 0
        ret = astar_path(
            node.character,
            node.location.name,
            (destx, desty),
            lambda a, b: sqrt((a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2),
        )
        node.engine.debug(
            f"{node.name}'s shortest path to {destx, desty} is {ret}"
        )
        return ret

    @phys.rule
    def go_places(char):
        from time import monotonic
        from networkx.exception import NetworkXNoPath

        def log_as_completed(fut):
            try:
                char.engine.debug(
                    f"Got path for {fut.thing.name}: {fut.result()}"
                )
            except NetworkXNoPath:
                char.engine.debug(f"No path for {fut.thing.name}")

        futs = []
        for thing in char.thing.values():
            fut = char.engine.submit(
                char.engine.function.find_path_somewhere, thing
            )
            fut.thing = thing
            fut.add_done_callback(log_as_completed)
            futs.append(fut)
        with char.engine.batch():
            for fut in futs:
                try:
                    result = fut.result()
                    thing = fut.thing
                    start = monotonic()
                    thing.follow_path(result, check=False)
                    char.engine.debug(
                        f"followed path for thing {thing.name} in {monotonic() - start:.2} seconds"
                    )
                except NetworkXNoPath:
                    char.engine.debug(
                        f"got no path for thing {fut.thing.name}"
                    )
                    continue

    @go_places.trigger
    def turn_one_only(char):
        return char.engine.turn == 1