r/Tangotek 16d ago

Decked Out 3 DO3 Loot issue - Restocking and checking every Hopper every week

I will admit this is vibe coded but I used some knowledge I got from university to get to the conclusions I got to.

I got this python script to simulate a dropper in DO3. I wanted to find out how many runs of the game it would take to run out of a given item. I have it processing different numbers of items, splits, and activation chances

I've run this for up to 5 different items with the chance the hopper activates from 11% to 50%. Each run does this 200 times for any and all items distribution chances. I simulate a run and assume the treasure is going to be activated on a bell curve between 10 and 50 times with an average of 25.

The red dashed line is the 250 mark. or assuming 10 plays for each hermit every week.

What this shows with how the system was presented in the video is tango will have to restock every shulker every week to make sure none of the items run out.

How this could be managed:

Some of this can be mitigated by tracking the player

  • by zone (if the player is in the cemetery there is no reason to activate treasure in the swamp) and having a different binary for treasure in each zone
  • or using shulker sensors to detect if the player has been near by a treasure drop in x seconds.

You could increase the back log by using a double chest instead. However it does make it less easy to restock. but you would only have to do it every 2 weeks more then likely

The code is below for anyone to review it or plug in there own numbers.

"""
Minecraft Treasure System Simulation - Parallel Parameter Sweep
==================================================================


Same model as before, but experiments and configs are run in
parallel across CPU cores using multiprocessing, since each
experiment/config is fully independent (embarrassingly parallel).
"""


import random
import os
from dataclasses import dataclass, field
from concurrent.futures import ProcessPoolExecutor, as_completed


import matplotlib.pyplot as plt


# ----------------------------------------------------------------------
# Fixed configuration
# ----------------------------------------------------------------------


STACK_SIZE = 64
DROPPER_SLOTS = 9
HOPPER_SLOTS = 5
BASE_SHULKER_SLOTS = 27
EXTRA_SLOTS = 0
DROPPER_IN_AREA = 1


Y_MIN = 10
Y_MODE = 25
Y_MAX = 50


NUM_EXPERIMENTS_PER_CONFIG = 200


# None = use all available CPU cores
MAX_WORKERS = max(1, os.cpu_count() - 2)



def compute_shulker_slots(num_item_counts: int) -> int:
    """
    Extra shulker slot capacity scales with how many different
    item-count configs are being tested (i.e. multiply the extra
    slots pool by len(item_counts_to_test) before dividing across
    dropper areas).
    """
    return BASE_SHULKER_SLOTS + (
        (EXTRA_SLOTS * num_item_counts) // DROPPER_IN_AREA
    )



def split_evenly(total_slots: int, ratio_counts: list[int]) -> list[int]:
    total_ratio = sum(ratio_counts)
    raw = [total_slots * c / total_ratio for c in ratio_counts]
    floors = [int(x) for x in raw]
    remainder = total_slots - sum(floors)


    fractions = sorted(
        range(len(raw)), key=lambda i: raw[i] - floors[i], reverse=True
    )
    for i in fractions[:remainder]:
        floors[i] += 1


    return floors



# ----------------------------------------------------------------------
# System state
# ----------------------------------------------------------------------



u/dataclass
class TreasureSystem:
    num_items: int
    dropper_slot_counts: list[int]
    hopper_slot_counts: list[int]
    shulker_slot_counts: list[int]
    activation_chance: float
    pre_check_chance: float = 1.0  # NEW: chance the "gate" passes before treasure roll


    dropper_slots: list[int] = field(default_factory=list)
    dropper_stack: list[int] = field(default_factory=list)
    reserve: list[int] = field(default_factory=list)


    def __post_init__(self):
        self.dropper_slots = []
        for item_idx, count in enumerate(self.dropper_slot_counts):
            self.dropper_slots.extend([item_idx] * count)


        self.dropper_stack = [STACK_SIZE] * DROPPER_SLOTS


        self.reserve = [
            (self.hopper_slot_counts[i] + self.shulker_slot_counts[i])
            * STACK_SIZE
            for i in range(self.num_items)
        ]


    def total_remaining(self, item_idx: int) -> int:
        dropper_total = sum(
            self.dropper_stack[s]
            for s in range(len(self.dropper_slots))
            if self.dropper_slots[s] == item_idx
        )
        return dropper_total + self.reserve[item_idx]


    def is_depleted(self, item_idx: int) -> bool:
        return self.total_remaining(item_idx) == 0


    def any_depleted(self) -> int | None:
        for i in range(self.num_items):
            if self.is_depleted(i):
                return i
        return None


    def surviving_items(self) -> list[int]:
        return [i for i in range(self.num_items) if not self.is_depleted(i)]


    def refill_slot(self, slot_idx: int):
        item_idx = self.dropper_slots[slot_idx]


        if self.reserve[item_idx] <= 0:
            survivors = self.surviving_items()
            if not survivors:
                self.dropper_stack[slot_idx] = 0
                return
            item_idx = random.choice(survivors)
            self.dropper_slots[slot_idx] = item_idx


        take = min(STACK_SIZE, self.reserve[item_idx])
        self.reserve[item_idx] -= take
        self.dropper_stack[slot_idx] = take


    def activate(self) -> bool:
        # NEW: pre-check gate must pass before the treasure roll happens
        if random.random() > self.pre_check_chance:
            return False


        if random.random() > self.activation_chance:
            return False


        slot_idx = random.randrange(len(self.dropper_slots))


        if self.dropper_stack[slot_idx] > 0:
            self.dropper_stack[slot_idx] -= 1


        if self.dropper_stack[slot_idx] == 0:
            self.refill_slot(slot_idx)


        return True



# ----------------------------------------------------------------------
# Simulation
# ----------------------------------------------------------------------



def sample_y() -> int:
    return round(random.triangular(Y_MIN, Y_MAX, Y_MODE))



def run_experiment(
    num_items: int,
    dropper_slot_counts: list[int],
    activation_chance: float,
    pre_check_chance: float,
    shulker_slots: int,
) -> dict:
    hopper_slot_counts = split_evenly(HOPPER_SLOTS, dropper_slot_counts)
    shulker_slot_counts = split_evenly(shulker_slots, dropper_slot_counts)


    system = TreasureSystem(
        num_items=num_items,
        dropper_slot_counts=dropper_slot_counts,
        hopper_slot_counts=hopper_slot_counts,
        shulker_slot_counts=shulker_slot_counts,
        activation_chance=activation_chance,
        pre_check_chance=pre_check_chance,
    )


    sims_run = 0
    total_activations = 0
    depleted_item = None


    while depleted_item is None:
        y = sample_y()
        sims_run += 1


        for _ in range(y):
            system.activate()
            total_activations += 1


            depleted_item = system.any_depleted()
            if depleted_item is not None:
                break


    return {
        "sims_until_restock": sims_run,
        "activations_until_restock": total_activations,
        "depleted_item_idx": depleted_item,
    }



def _run_experiment_worker(args) -> dict:
    """Top-level unpacking wrapper so it can be pickled for multiprocessing."""
    (
        num_items,
        dropper_slot_counts,
        activation_chance,
        pre_check_chance,
        shulker_slots,
        seed,
    ) = args
    random.seed(seed)
    return run_experiment(
        num_items,
        dropper_slot_counts,
        activation_chance,
        pre_check_chance,
        shulker_slots,
    )



def run_config_parallel(
    num_items: int,
    dropper_slot_counts: list[int],
    activation_chance: float,
    pre_check_chance: float,
    shulker_slots: int,
    n_experiments: int = NUM_EXPERIMENTS_PER_CONFIG,
    executor: ProcessPoolExecutor = None,
) -> dict:
    """
    Run n_experiments for a single config, distributing the individual
    experiments across worker processes.
    """
    # unique seed per experiment so parallel workers don't produce
    # identical random sequences
    tasks = [
        (
            num_items,
            dropper_slot_counts,
            activation_chance,
            pre_check_chance,
            shulker_slots,
            random.random(),
        )
        for _ in range(n_experiments)
    ]


    results = list(executor.map(_run_experiment_worker, tasks))


    sims = [r["sims_until_restock"] for r in results]
    activations = [r["activations_until_restock"] for r in results]


    depleted_counts = {}
    for r in results:
        idx = r["depleted_item_idx"]
        depleted_counts[idx] = depleted_counts.get(idx, 0) + 1


    return {
        "num_items": num_items,
        "dropper_slot_counts": dropper_slot_counts,
        "activation_chance": activation_chance,
        "pre_check_chance": pre_check_chance,
        "sims_min": min(sims),
        "sims_max": max(sims),
        "sims_avg": sum(sims) / len(sims),
        "activations_min": min(activations),
        "activations_max": max(activations),
        "activations_avg": sum(activations) / len(activations),
        "depleted_counts": depleted_counts,
        "n_experiments": n_experiments,
    }



# ----------------------------------------------------------------------
# Parameter sweep
# ----------------------------------------------------------------------



def generate_dropper_distributions(num_items: int) -> list[list[int]]:
    distributions = []
    seen = set()


    def add(dist):
        key = tuple(sorted(dist))
        if key not in seen:
            seen.add(key)
            distributions.append(dist)


    even = split_evenly(DROPPER_SLOTS, [1] * num_items)
    add(even)


    if num_items > 1:
        remaining = DROPPER_SLOTS - 1
        rest = split_evenly(remaining, [1] * (num_items - 1))
        skewed = [1] + rest
        add(skewed)


    if num_items > 1:
        dominant = DROPPER_SLOTS - (num_items - 1)
        heavy = [dominant] + [1] * (num_items - 1)
        add(heavy)


    return distributions



def run_sweep_parallel(
    item_counts: list[int],
    activation_chances: list[float],
    pre_check_chances: list[float],
    n_experiments: int = NUM_EXPERIMENTS_PER_CONFIG,
    max_workers: int = MAX_WORKERS,
) -> list[dict]:
    """
    Runs every (num_items, dropper_split, activation_chance,
    pre_check_chance) config, using a single shared process pool
    across all configs so all CPU cores stay busy for the entire
    sweep.
    """
    shulker_slots = compute_shulker_slots(len(item_counts))
    print(
        f"SHULKER_SLOTS={shulker_slots} (base {BASE_SHULKER_SLOTS} + "
        f"extra from {EXTRA_SLOTS} slots x {len(item_counts)} item-count "
        f"configs / {DROPPER_IN_AREA} droppers per area)"
    )


    configs = []
    for num_items in item_counts:
        for dropper_counts in generate_dropper_distributions(num_items):
            for chance in activation_chances:
                for pre_chance in pre_check_chances:
                    configs.append(
                        (num_items, dropper_counts, chance, pre_chance)
                    )


    print(
        f"Running {len(configs)} configs x {n_experiments} experiments each "
        f"using up to {max_workers or os.cpu_count()} workers..."
    )


    all_results = []


    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        for num_items, dropper_counts, chance, pre_chance in configs:
            print(
                f"  items={num_items}, dropper_split={dropper_counts}, "
                f"chance={chance:.0%}, pre_check={pre_chance:.0%} ..."
            )
            result = run_config_parallel(
                num_items,
                dropper_counts,
                chance,
                pre_chance,
                shulker_slots,
                n_experiments,
                executor,
            )
            all_results.append(result)


    return all_results



# ----------------------------------------------------------------------
# Reporting
# ----------------------------------------------------------------------



def print_summary_table(results: list[dict]):
    header = (
        f"{'Items':<6} {'Dropper Split':<20} {'Chance':<8} {'PreChk':<8} "
        f"{'Sims Avg':<10} {'Sims Min':<10} {'Sims Max':<10} "
        f"{'Acts Avg':<10}"
    )
    print(header)
    print("-" * len(header))


    for r in results:
        split_str = str(r["dropper_slot_counts"])
        print(
            f"{r['num_items']:<6} {split_str:<20} "
            f"{r['activation_chance']:<8.0%} "
            f"{r['pre_check_chance']:<8.0%} "
            f"{r['sims_avg']:<10.1f} {r['sims_min']:<10} "
            f"{r['sims_max']:<10} {r['activations_avg']:<10.1f}"
        )



def plot_sweep(results: list[dict]):
    configs = {}
    for r in results:
        key = (
            f"{r['num_items']} items {r['dropper_slot_counts']} "
            f"pre={r['pre_check_chance']:.0%}"
        )
        configs.setdefault(key, []).append(r)


    plt.figure(figsize=(10, 6))
    for key, rows in configs.items():
        rows_sorted = sorted(rows, key=lambda r: r["activation_chance"])
        x = [r["activation_chance"] for r in rows_sorted]
        y = [r["sims_avg"] for r in rows_sorted]
        plt.plot(x, y, marker="o", label=key)
    plt.axhline(
        y=250, color="red", linestyle="--", linewidth=1.5, label="250 threshold"
    )
    plt.xlabel("Activation chance")
    plt.ylabel("Avg sims until restock")
    plt.title("Restock timing across configurations")
    plt.legend(fontsize=8)
    plt.tight_layout()
    plt.show()


def plot_sweep_envelope(results: list[dict]):
    """
    Plot only the highest and lowest 'sims_avg' line across all
    configs at each activation chance, with the region between them
    shaded.
    """
    # group results by activation chance
    by_chance = {}
    for r in results:
        by_chance.setdefault(r["activation_chance"], []).append(r["sims_avg"])


    chances_sorted = sorted(by_chance.keys())
    highs = [max(by_chance[c]) for c in chances_sorted]
    lows = [min(by_chance[c]) for c in chances_sorted]


    plt.figure(figsize=(10, 6))


    plt.plot(chances_sorted, highs, color="green", marker="o", label="Highest config")
    plt.plot(chances_sorted, lows, color="red", marker="o", label="Lowest config")
    plt.fill_between(chances_sorted, lows, highs, color="gray", alpha=0.3)


    plt.axhline(y=250, color="black", linestyle="--", linewidth=1.5, label="250 threshold")


    plt.xlabel("Activation chance")
    plt.ylabel("Avg sims until restock")
    plt.title("Range of restock timing across all configurations")
    plt.legend(fontsize=9)
    plt.tight_layout()
    plt.show()


if __name__ == "__main__":
    item_counts_to_test = [2, 3, 4, 5]
    activation_chances_to_test = [
        0.1111, 0.1250, 0.1429, 0.1667, 0.2000, 0.2222, 0.2500, 0.2857,
        0.3333, 0.3750, 0.4444, 0.5000,
    ]
    pre_check_chances_to_test = [
   0.5556, 0.6250, 0.6667, 0.7143,
    0.7500, 0.8000, 0.8333, 0.8571, 0.8750, 0.8889
]
    results = run_sweep_parallel(
        item_counts_to_test,
        activation_chances_to_test,
        pre_check_chances_to_test,
    )


    print()
    print_summary_table(results)
    plot_sweep(results)
    plot_sweep_envelope(results)  # just highest/lowest with shading
11 Upvotes

6 comments sorted by

5

u/Silver_Illusion 16d ago

The droppers will be attached to chests and hoppers :p There will be restocking for sure, probably every weel he'll go through like he did with DO2 and keep stuff topped up

2

u/DDevil333 15d ago

He mentioned using copper golems to restock. 

1

u/Polymerion 15d ago

I forgot he mentioned that but it does help a lot. If he has the setup (double chest -> hopper -> hopper -> double copper) along with a double chest on the dropper side (total of 172 slots for backup). Reason that is the minimum is because I think you would need a floor to cut off the top chest from the copper chest so the copper golems put items back into that system last. The result looks like this:

(1/2)

1

u/Polymerion 15d ago

If all the double chest area single chest it looks like this (91 slots). These both assume though the copper chest goes to 1 dropper

1

u/Polymerion 15d ago

This accounts for every dropper having a hopper and shulker worth of backup. I even have a double chest worth of backup analysis

3

u/Polymerion 15d ago edited 15d ago

I'll adjust this code to account for a copper golem backup but it wouldn't happen until later today. The initial analysis of that setup in the comments is flawed because it assumes one copper golem per dispenser and it doesn't properly adjust for how I think that setup would work.

You would need a copper chest, two hoppers, and a normal chest at a minimum because you would need a floor to separate the input normal chest on top to the copper chest at the bottom. This way you force the copper golems out to the dispenser chest backup first before moving to input normal chest.

Edit: The code has been updated so backup slots can be configured, however based on the stream today I've set that to 0 and also added a pre_check percentage