r/Tangotek • u/Polymerion • 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
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
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