Script to find all spawners and groups them as double, triple, quadruple etc

spawners.py

import anvil
import datetime
import itertools
import os.path as path

# Setting this to 1 turns off multi threading functionality
threadCount = 4

regionFilesDirectory = 'server/world_old/region'

maximumDistanceBetweenSpawners = 32

#csv_filename = "./spawners/spawnerdata" + str(datetime.datetime.now().strftime("%Y%m%d-%H%M%S")) + ".csv"

class WorldLoader:

    def __init__(self, constraints):
        self.chunkconstraints = [(constraints[0] >> 4), (constraints[1] >> 4), (constraints[2] >> 4),
                            (constraints[3] >> 4)]
        self.region = None

    def reduceRegion(self, regionX, regionZ, function, initializer=None):
        result = initializer
        if initializer is None:
            result = []
        region = self.region
        if region is None:
            region = anvil.Region.from_file(path.join(regionFilesDirectory, f'r.{regionX}.{regionZ}.mca'))
            self.region = region

        constraints = self.chunkconstraints

        xStart = max((regionX << 5) + 0, constraints[0])
        xStop = min((regionX << 5) + 32, constraints[2])

        zStart = max((regionZ << 5) + 0, constraints[1])
        zStop = min((regionZ << 5) + 32, constraints[3])

        for x in range(xStart, xStop + 1):
            for z in range(zStart, zStop + 1):
                try:
                    chunk = anvil.Chunk.from_region(region, x, z)
                    result = function(result, chunk)
                except Exception as e:
                    print("Failed to load chunk: ", x, z)
                    raise e
        return result


def split_area_into_regions(mining_rectangle):
    mining_regions = []
    for i in range((mining_rectangle[0]) >> 9, ((mining_rectangle[2]) >> 9) + 1):
        for j in range((mining_rectangle[1]) >> 9, (mining_rectangle[3] >> 9) + 1):
            mining_regions.append([max(i << 9, mining_rectangle[0]), max((j << 9), mining_rectangle[1]),
                                   min(((i + 1) << 9) - 1, mining_rectangle[2]),
                                   min(((j + 1) << 9) - 1, mining_rectangle[3])])
    return mining_regions


class Spawner:

    def __init__(self, x, y, z, mob):
        self.coordinates = (x, y, z)
        self.x = x
        self.y = y
        self.z = z
        if "minecraft:" in mob:
            self.mob = mob[10:]
        else:
            self.mob = mob

    def __hash__(self):
        return hash(self.coordinates)

    def __eq__(self, other):
        return self.coordinates == other.coordinates

    def __str__(self):
        return str(self.x) + "," + str(self.y) + "," + str(self.z)

    def __repr__(self):
        return "Spawner(" + str(self.x) + ", " + str(self.y) + ", " + str(self.z) + ', "' + self.mob + '")'

class SpawnerGroup:

    def __init__(self, spawners):
        self.set = set(spawners)
        self.tuple = tuple(sorted(spawners, key=lambda s: (s.x, s.y, s.z)))

    def __hash__(self):
        return hash(self.tuple)

    def __getitem__(self, item):
        return self.tuple[item]

    def __eq__(self, other):
        return self.set == other.set

    def __str__(self):
        return ",".join(map(str, self.tuple));

    def __repr__(self):
        return "SpawnerGroup(" + repr(self.tuple) + ")"

    def __iter__(self):
        return iter(self.set)

    def __contains__(self, item):
        return item in self.set

    def symmetric_difference(self, other):
        return self.set.symmetric_difference(other.set)

    def union(self, other):
        return SpawnerGroup(self.set.union(other.set))

def reducer(accumulator, current):
    for te in current.data['TileEntities']:
        if te["id"].value == "minecraft:mob_spawner":
            accumulator.append(Spawner (
                te["x"].value,
                te["y"].value,
                te["z"].value,
                te["SpawnData"]['id'].value
            ))
    return accumulator

def simulateForRegion(mining_area):
    loader = WorldLoader(mining_area)
    regionX = mining_area[0] >> 9
    regionZ = mining_area[1] >> 9
    print(" ".join(("Scanning: ", str(mining_area), " Region: (", str(regionX), ",", str(regionZ), ")")))
    result = loader.reduceRegion(regionX, regionZ, reducer)
    return result

def process_work(todo, resultsQueue):
    while True:
        task = todo.get();
        try:
            result = simulateForRegion(task)
            resultsQueue.put(result)
        except Exception as e:
            import traceback
            import sys

            print(type(e), e)
            print(task)
            traceback.print_exc(file=sys.stderr)
            resultsQueue.put({})

class Executor:

    def __init__(self):
        self.processes = []
        self.workCount = 0
        self.results = []
        self.aggregatedResult = None

        if threadCount > 1:
            from multiprocessing import Process, Queue
            self.toDo = Queue()
            self.resultsQueue = Queue()

            for i in range(threadCount):
                p = Process(target=process_work, args=(self.toDo, self.resultsQueue))
                self.processes.append(p)
                p.start()

    def execute(self, task):
        if threadCount > 1:
            self.workCount += 1
            self.toDo.put(task)
        else:
            self.results.extend(simulateForRegion(task))


    def getResults(self, start):
        if self.aggregatedResult is None:
            if threadCount > 1:
                for i in range(1, self.workCount + 1):
                    current = self.resultsQueue.get()

                    self.results.extend(current)
                    time = datetime.datetime.now() - start
                    progress = i / self.workCount
                    print(" ".join((str(i), "of", str(self.workCount), "region files done.", str(100 * (progress)), "%",
                          "Elapsed time:", str(time), "Remaining:", str((time * (1 - progress) / progress)))))
                for process in self.processes:
                    process.terminate()
            self.aggregatedResult = self.results
        return self.aggregatedResult

def distance(o1, o2):
    return (abs(o1.x - o2.x) ** 2 + abs(o1.y - o2.y) ** 2 + abs(o1.z - o2.z) ** 2) ** 0.5

# The generic algorithm for finding close spawners is as follows.
# To find all groups of n spawners:
#     iterate all combinations of 2 of all groups of n - 1 spawners  (i.e. combinations(groupNminus1, 2))
#         Find how many spawners are in both groups and how many are only in one group
#         if there are n-2 spawners that are in both groups and 2 spawners that are only in one of the two groups:
#             measure distances between each of the two spawners and each spawner in the group that it is not in
#             if all the distances are small enough:
#                 Create a new group containing the n spawners from combining the two groups of n-1
#                 Add this new group to the set containing all groups of n spawners (use a set so that it removes duplicates)
def groupCloseSpawners(spawners):
    if len(spawners[1]) <= 1:
        spawners['max'] = len(spawners)
        return

    counter = 2
    while 1:
        previous = spawners[counter - 1]
        the_next = set()
        to_remove = set()
        for c1, c2 in itertools.combinations(previous, 2):
            # The spawners that are only in one of the two groups
            # If this is 2 then there is no need to count the ones that are in both groups
            diff = c1.symmetric_difference(c2)
            if len(diff) != 2:
                continue
            is_valid = True
            for different in diff:
                other = c1
                if different in c1:
                    other = c2
                is_valid = is_valid and all(
                    map(
                        lambda spawner: distance(different, spawner) < maximumDistanceBetweenSpawners,
                        other))
                if not is_valid:
                    break
            if is_valid:
                the_next.add(c1.union(c2))
                to_remove.add(c1)
                to_remove.add(c2)

        # Remove spawners that have been grouped from the previous group in order to not show them in the results
        # For example if you have 3 spawners and two of them are close enough to each other
        # Report that there is 1 single spanwer and 1 double spanwer instead of 3 single and 1 double
        for spawnerGroup in to_remove:
            if spawnerGroup in previous:
                previous.remove(spawnerGroup)

        if len(the_next) > 0:
            spawners['max'] = counter
            spawners[counter] = the_next
        else:
            spawners['max'] = counter - 1
        if len(the_next) <= 1:
            break

        counter = counter + 1


def findCloseSpawnersGroupByMob(spawners):
    key = lambda x: x.mob
    allSpawnersGroupedByMob = {}
    allSpawnersSet = set()
    # Group spawners by mob type
    for mob, mobSpawners in itertools.groupby(spawners, key=key):
        mobDict = allSpawnersGroupedByMob.setdefault(mob, dict())
        mobSingleSet = mobDict.setdefault(1, set())
        for spawner in mobSpawners:
            # Treat each spawner as it's own group of 1 spawners
            # Not very efficient, but makes it simpler to implement the general algorithm
            sg = SpawnerGroup((spawner,))
            mobSingleSet.add(sg)
            allSpawnersSet.add(sg)

    for mob in allSpawnersGroupedByMob:

        mobSpawners = allSpawnersGroupedByMob[mob]
        # Group spawners by position
        groupCloseSpawners(mobSpawners)

        print(mob, ":", sep="")

        # Here is good place to filter out mob types. For example:
        #if mob in ("zombie", "cave_spider"):
        #    continue
        # This will skip printing out spawners for zombies and cave spiders
        # To reverse it add a 'not', like 'if not mob in' (or 'if mob not in')

        for numberOfSpawnersInGroup in range(1, mobSpawners['max'] + 1):
            print("\t", len(mobSpawners[numberOfSpawnersInGroup]), "groups of", numberOfSpawnersInGroup, "spawners")

            # Here you can filter out based on number of spawners in the group
            #if numberOfSpawnersInGroup < 2:
            #   continue

            #Print first spawner of each group
            #print(" & ".join(map(lambda x: str(x[0]), mobSpawners[numberOfSpawnersInGroup])))

            # Print out the actual groups of spawners
            #for spawnerGroup in mobSpawners[numberOfSpawnersInGroup]:
            #    print("\t\t", spawnerGroup)

    allSpawners = { 1 : allSpawnersSet }
    groupCloseSpawners(allSpawners)
    print("Combinations of spawners of any type:")
    for numberOfSpawnersInGroup in range(1, allSpawners['max'] + 1):
        listOfSpawnerGroups = allSpawners[numberOfSpawnersInGroup]

        # Here you can filter out based on number of spawners in the group
        #if numberOfSpawnersInGroup < 3:
        #    continue

        # Filter out all groups that have only one type of spawner as those are reported above in the per mob section
        listOfSpawnerGroups = list(
            filter(
                lambda spawnerGroup:
                    not all(
                        map(
                            lambda spawner: spawnerGroup[0].mob == spawner.mob,
                            spawnerGroup)),
                listOfSpawnerGroups))

        if len(listOfSpawnerGroups) > 0:
            print("\t", len(listOfSpawnerGroups), "groups of", numberOfSpawnersInGroup, "spawners")


            #Print out the actual groups of spawners
            #for spawnerGroup in listOfSpawnerGroups:
            #    print("\t\t", str(spawnerGroup))

def simulate(area, start):
    regions = split_area_into_regions(area)

    executor = Executor()

    for region in regions:
        executor.execute(region)

    key = lambda x: x.mob
    results = sorted(executor.getResults(start), key = key);

    #print out the raw data in a for that is easy to use as python input
    #print("[", end = "")
    #for r in results:
    #    print(repr(r) + ",")
    #print("]")

    findCloseSpawnersGroupByMob(results)

def find_spawners():
    print("Starting simulation")
    area = [-2048, -2048, 2047, 2047]
    #area = [-2048, -2048, -1537, -1537]
    start = datetime.datetime.now()
    simulate(area, start)
    end = datetime.datetime.now()
    print("Finished simulation in ", end - start)

if __name__ == "__main__":
    find_spawners()
    pass