SolvedHow-toUpdated
The Farmer Was Replaced: Decimalist's Code
Short answer
Collection of my code. Includes: Parallelization --- Cactus: Shakersort --- Sunflower: Buckets --- Pumpkin: Queue --- Polyculture: random Chain or special-crop --- Maze: Multidrone Cheese
Steam Community guide by decimalist.
Intro
Steam guide for game version V1.0 updated as of 2025 Oct 19
- This steam guide's thumbnail is completely unrelated. I just chose a funny square image that I had on my desktop. I love that meme.
- I might have missed some steam text formatting such as [i] in-code becoming italics. Use your best judgement!
- I'm enjoying the Oct 10 V1.0 update very much. I'm just uploading what I have currently.
- I don't really have any automation routines for you nor any meta-optimization algorithms for you. Just basic code.
- I follow github etiquette. My code is your code, and your code is my code, and we all steal are inspired off of each other.
- I'm just a mediocre coder. Comment any fixes or improvements for me to steal for readers to take note of.
- Unfortunately I'm also not a beginner to programming. I won't be able to help you if you don't already know some semblance of Pythonic or pseudocoding.
- Lastly, I'm guessing that <code> in this game is Pythonic custom DSL. A lot of syntactic sugar is missing, such as:
- using backslashes to indicate a continued line of code
- list comprehensions
- nonlocal keyword
- yield keyword and generator functions
- pointer passing (eg. **args)
- etc. I'm sure you'll find some missing too
Notes
TODO list for me
- DONE **args / Parallelism for non Axis-Aligned (AA) subtasks/drone-jobs using nonzero-arg subtasks which can handle inputs defined by Factory functions
- DONE Pumpkins parallelization
- DONE Polyculture parallelization
- Figure out how to detect the first-finished multiprocess drone job for the "long" parallelization strategy, instead of naively pulling the first-started drone job. This will improve speed for "long" strategy if the drone task does not have uniform job length / time taken.
- "finding" code for other parts of the game
Wish list to dev
- using "== None" ticks my brain, years "is None" down the drain
- Try/Exception
- Using blackslash to indicate continue on newline of code
- List comprehensions
- yield keyword and generator functions
- Immediate iterable decoupling for dicts (for k, v in dict:)
- global keyword accepts multiple variables at once
- OOP
Additional Notes
- The game uses wraparound movement. Treat the game board as an infinite tiling or a torus.
- My coding style is to be "generally safe", which includes stuff like function parameter checking, etc. This introduces additional if statements and code for the drone to slog through, so if you really care about speeding up code by a couple ticks I recommend you make a copy and remove those checks. If you want to use code for leaderboard running please do this.
- Whenever I use coordinates, I always use one argument (x, y) instead of two separate args x and y. It's a matter of personal preference I guess. (Also the function get_companion() returns the same type, a coordinate 2-tuple)
- Disclaimer: For multiprocessing/parallelization code, I only ever coded those with worldsize and n_drones = 32. I haven't really tested the other cases such as multidrone not unlocked, or weird drone-worldsize combinations.
Let us proceed fellow farmers.
Helpers
utils_num
def is_even(num):
return num % 2 == 0
def is_odd(num):
return not is_even(num)
def clamp(num, left, right):
Clamp a number between a given min and max.
return min(max(num, left), right)
def ring_clamp(num, width):
return num % width
utils_list
I commonly use a 1d list to keep track of information on the game board. 2d matrix is too complicated for the scope so I just flatten it. You'll see me call this "serialized" version or something.
def all(ls):
for x in ls:
if not x:
return False
return True
def any(ls):
for x in ls:
if x:
return True
return False
def index(ls, item):
for i in range(len(ls)):
if ls[i] == item:
return i
return None # TODO raise exception
def max_with_index(ls):
Args:
ls: list
Returns:
value: maximum item
index: maximum index
if not ls:
return None, -1
max_val = ls[0]
max_idx = 0
for i in range(1, len(ls)):
cur_val = ls[i]
if cur_val > max_val:
max_val = cur_val
max_idx = i
return max_val, max_idx
def extend(ls1, ls2):
Extend ls1 with ls2's values.
for x in ls2:
ls1.append(x)
def reversed(ls):
Reverse the list.
res = []
for i in range(len(ls) -1, -1, -1):
res.append(ls[i])
return res
def enumerate_list(ls):
Returns a list of (idx, val) pairs
i = 0
result = []
for item in ls:
result.append((i, item))
i = i + 1
return result
def serialize_2d_index(idx, dim_len):
Args:
idx: 2-tuple
dim_len: int length of 0th dimension
Returns:
int: serialized index
return idx[1] * dim_len + idx[0]
def deserialize_2d_index(idx, dim_len):
Args:
idx: int
dim_len: int length of 0th dimension
Returns:
2-tuple: 2d matrix index
return idx % dim_len, idx // dim_len
def serialize_2d_matrix(m):
Flattens a 2d matrix into a list.
Args:
m: List[List]
Returns:
list: flattened 1d matrix
res = []
for i in range(len(m)):
extend(res, m[i])
return res
def deserialize_2d_matrix_CC(m, dim_len):
Rebuilds 2d matrix from 1d.
Uses the Contiguous Chunks strategy.
Args:
m: List
dim_len: int 0th dim target length (number of rows)
Returns:
list[list]: 2d matrix or None if invalid
if dim_len <= 0:
return None
n = len(m)
if n % dim_len != 0:
return None # must divide evenly for strict reshape
build dim_len rows
cols = n // dim_len
res = []
i = 0
while i < dim_len:
res.append([])
i = i + 1
fill row-major (contiguous chunks)
i = 0
while i < n:
row_idx = i // cols
res[row_idx].append(m[i])
i = i + 1
return res
def deserialize_2d_matrix_unsafe_CC(m, dim_len):
Rebuilds a 2D matrix from a 1D list,
distributing items as evenly as possible.
Uses the Contiguous Chunks strategy.
Args:
m: list
dim_len: int 0th dim target length (number of rows)
Returns:
list[list]: 2D "best-effort" matrix distributed CC
if dim_len <= 0:
return [m]
n = len(m)
if n == 0:
temp = []
for _ in range(dim_len):
temp.append([])
return temp
Compute base size and remainder
(extra elements to distribute)
rows = dim_len
base = n // rows
remainder = n % rows
res = []
idx = 0
i = 0
while i < rows:
Give one extra element to
the first remainder rows
if i < remainder:
extra = 1
else:
extra = 0
row_len = base + extra
row = m[idx:idx + row_len]
res.append(row)
idx += row_len
i += 1
If some elements remain (shouldn't
happen), append them all to
the last row
if idx < n:
res[-1].extend(m[index:])
return res
def deserialize_2d_matrix_RR(m, dim_len):
Rebuilds 2d matrix from 1d.
Uses the Round Robin strategy.
Args:
m: List
dim_len: int 0th dim target length (number of rows)
Returns:
list[list]: 2d matrix or None if invalid
if dim_len <= 0:
return None
NOTE update when list comps
res = []
i = 0
while i < dim_len:
res.append([])
i = i + 1
n = len(m)
if n == 0:
return res
if n % dim_len != 0:
return None # must divide evenly for strict reshape
build exactly dim_len rows
res = []
i = 0
while i < dim_len:
res.append([])
i = i + 1
round robin placement
for idx, item in enumerate_list(m):
res[idx % dim_len].append(item)
return res
def deserialize_2d_matrix_unsafe_RR(m, dim_len):
Rebuilds a 2D matrix from a 1D list,
distributing items as evenly as possible.
Uses the Round Robin strategy.
Args:
m: list
dim_len: int 0th dim target length (number of rows)
Returns:
list[list]: 2D "best-effort" matrix distributed RR
if dim_len <= 0:
return [m]
n = len(m)
NOTE update when list comps [[] for _ in range(dim_len)]
res = []
for _ in range(dim_len):
res.append([])
if n == 0:
return res
round robin placement
for idx, item in enumerate_list(m):
res[idx % dim_len].append(item)
return res
utils_move
For some reason, you can't import "hidden" functions (names starting with underscore) like with Python. Thus my "_goto_y()" will just be "goto_y()".
from utils_num import *
from utils_list import *
from utils_drone import *
def get_pos():
return (get_pos_x(), get_pos_y())
def _shortest_delta(curr, dest, size):
d = (dest - curr) % size
if d > size / 2:
d -= size
return d
def goto_x(tx):
ws = get_world_size()
cx = get_pos_x()
dx = _shortest_delta(cx, tx, ws)
if dx > 0:
for _ in range(dx):
move(East)
elif dx < 0:
for _ in range(-dx):
move(West)
return True
def goto_y(ty):
ws = get_world_size()
cy = get_pos_y()
dy = _shortest_delta(cy, ty, ws)
if dy > 0:
for _ in range(dy):
move(North)
elif dy < 0:
for _ in range(-dy):
move(South)
return True
def goto(target):
The drone makes use of wraparound.
Args:
target: Tuple of length 2
Returns:
bool: Success state
tx, ty = target
goto_x(tx)
goto_y(ty)
return True
def u_turn(dir):
Given a direction, return the
opposite direction
if dir == North:
return South
elif dir == South:
return North
elif dir == East:
return West
elif dir == West:
return East
else:
return None
def calculate_p_AA_positions(axis):
Calculates a list of all positions
across an axis, ordered farthest
distance first, inbetweens
alternating direction, and
closest distance last.
For the use of reaching the
farthest point first when starting
an axis aligned drone task.
Args:
axis: "x" or "y"
Returns:
list[int]
recompute_ws()
positions = []
if axis == "x":
temp_pos = get_pos_x()
else:
temp_pos = get_pos_y()
farthest = temp_pos - (ws // 2)
if farthest < 0:
farthest += ws
for i in range(1, ws+1):
if is_odd(i):
d = i // 2
else:
d = - i // 2
positions.append(ring_clamp(farthest + d, ws))
return positions
utils_item
def check_has_item(item, num=1):
DEPRECATED use can_pay_for()
Args:
item: obj under class Items
num: number to check
if num_items(item) < num:
return False
return True
def can_pay_for(entity, num=1):
Checks if I can pay for the
desired item(s).
costs = get_cost(entity)
if not costs:
return True
for k in costs:
v = costs[k]
if not check_has_item(k, v * num):
return False
return True
Helpers Pt 2
utils_grow
from utils_item import *
from utils_num import *
Used for logic in polyculture
entity_growth_timing = { # Entity: tuple[float, float], in seconds
Entities.Apple: (0.2, 0.2),
Entities.Bush: (3.2, 4.8),
Entities.Cactus: (1.0, 1.0),
Entities.Carrot: (4.8, 7.2),
Entities.Dead_Pumpkin: (0, 0), # JIC
Entities.Dinosaur: (0.18, 0.22),
Entities.Grass: (0.5, 0.5),
Entities.Hedge: (0, 0), # JIC
Entities.Pumpkin: (0.2, 3.8),
Entities.Sunflower: (5.6, 8.4),
Entities.Treasure: (0, 0), # JIC
Entities.Tree: (5.6, 8.4),
}
entity_allowed_groundtype = { # Entity: tuple containing allowed ground types
Entities.Apple: (Grounds.Grassland, Grounds.Soil),
Entities.Bush: (Grounds.Grassland, Grounds.Soil),
Entities.Cactus: (Grounds.Soil),
Entities.Carrot: (Grounds.Soil),
Entities.Dead_Pumpkin: (Grounds.Soil), # assign as if pumpkin
Entities.Dinosaur: (Grounds.Grassland, Grounds.Soil),
Entities.Grass: (Grounds.Grassland, Grounds.Soil),
Entities.Hedge: (), # JIC.
Entities.Pumpkin: (Grounds.Soil),
Entities.Sunflower: (Grounds.Soil),
Entities.Treasure: (), # JIC.
Entities.Tree: (Grounds.Grassland, Grounds.Soil),
}
entity_needs_soil = {
Exclusively needs soil.
Entities.Cactus,
Entities.Carrot,
Entities.Pumpkin,
Entities.Sunflower,
}
entity_needs_grassland = set() # Exclusively needs grassland.
def calculate_actual_growth_time():
Account for water level: linear interpolation
which is z = val * (left*(1-y) + right*(y))
ent = get_entity_type()
if ent == None or ent not in entity_growth_timing:
return 0
return entity_growth_timing[ent][0] * (1 / (1 + 4 * get_water()))
def maintain_water(thresh):
Args:
thresh: float threshold to maintain above
Returns:
bool: Success state
if not check_has_item(Items.Water):
return False
thresh = clamp(thresh, 0, 1)
while get_water() < thresh:
use_item(Items.Water)
return True
def till_for_entity(ent):
Prepare the ground for planting.
Simpler version but less safe:
if get_ground_type() not in entity_allowed_groundtype[ent]:
till()
ground = get_ground_type()
if ent in entity_needs_soil and ground != Grounds.Soil:
till()
elif ent in entity_needs_grassland and ground != Grounds.Grassland:
till()
return True
utils_farm
from utils_drone import *
from utils_move import *
def _factory_till_row(args_tuple):
start_pos = args_tuple
def _subtask_till_row():
recompute_ws()
goto(start_pos)
for _ in range(ws):
if get_ground_type() != Grounds.Soil:
till()
move(East)
return True
return _subtask_till_row
def _subtask_harvest_eastwards():
recompute_ws()
for _ in range(ws):
harvest()
move(East)
return True
def till_all():
recompute_parallelism()
px = get_pos_x()
args = []
positions = calculate_p_AA_positions("y")
for i in range(ws):
args.append((px, positions[i]))
return drone_task(_factory_till_row, args)
def harvest_all():
Use in case you planted an expensive field of crops.
recompute_parallelism()
return drone_naive_AA_task(_subtask_harvest_eastwards, North)
def randcoords():
recompute_ws()
return ((random() * ws) // 1, (random() * ws) // 1)
def get_companion_saferand():
Returns a random coordinate if
get_companion() fails.
ret = (get_companion(),)
if ret[0] == None:
return None, randcoords()
else:
return ret[0][0], ret[0][1]
utils_tick
def naive_wait(ticks):
Wait for ticks ticks.
This function is unsafe.
This function has overhead (adjusted).
1: Evaluate ticks - 3
1: Start for loop
1: IDK
for _ in range(ticks - 3):
pass
utils_tick
def naive_wait(ticks):
Wait for ticks ticks.
This function is unsafe.
This function has overhead (adjusted).
1: Evaluate ticks - 3
1: Start for loop
1: IDK
for _ in range(ticks - 3):
pass
Parallelization
If you don't have any experience with parallelization specifically multiprocessing as a means to achieve parallelization, then this part will be quite hard. Thankfully for me, I'm in uni and have taken an OS course which covered scheduling, multiprocessing, related topics etc so at least I'm not hopeless. (What the hell is a semaphore haha)
I wrote most of my parallelization functions in hopes that they can be used independently of Megafarm unlock level. I'm not sure if it works though, I haven't tried to simulate it. (eg. Hopefully, prevents using the function max_drones() if megafarm hasn't been unlocked, as that will raise errors).
Disclaimer, Caution: I've only tried all my parallelization code on the "short" mode with worldsize = max_drones() = 32.
I finally have a solution for **args-type subtasks!!! I sincerely hope that the drone_task function can help, I spent a lot of time on that. The way I deal with it is to treat **args as a tuple object containing m arguments to pass in. Then any subtask should be defined as taking in one tuple parameter, and can access parameters through iterating the tuple. Doesn't this remind you of how to use the Stack when programming in Assembly? :D
Also, this is the one time I've unironically used the factory design pattern. Thank you to this game for making me write "clean" code. Examples for how to make factory functions can be found in my implementations for polyculture, sunflowers, etc.
utils_drone
A solution to **args is to design the subtask to accept a tuple parameter.
def max_drones_safe():
Requires having unlocked Auto Unlock.
if num_unlocked(Unlocks.Megafarm) < 1:
return 1
else:
return max_drones()
Drone parallelism
def choose_parallelism(ws):
Recompute and decide how the main drone participates in parallel work.
My definition of parallelism assumes the work that needs
parallelism needs so under extension on one dimension
of the game board.
Heuristic:
- If Megafarm is locked: no parallelism ("none").
- Prefer "short" (main drone is a worker) when:
- The field width divides evenly among all drones, or
- The total number of drones is small relative to worldsize.
- However this assumes subtasks complete in roughly uniform
time, and that the n_drones*200ticks time overhead nearing
all subtasks' completion is acceptable. It could cause idle
drones or imbalance if subtask duration is random.
- Prefer "long" (main drone only delegates tasks) when:
- Removing the main drone yields an even partition, or
- The division is cleaner (greater GCD) without the main.
- There is a job requirement to proactively respawn drones
or subtask length follows a non-uniform distrubution.
- But I haven't found a solution to computing the first
job-completed drone task to efficiently manage drones
before they become idle. (TODO)
This balances even workload distribution with minimal idle time.
Megafarm not unlocked → no parallelism
if get_cost(Unlocks.Megafarm) == get_cost(Unlocks.Megafarm, 0):
return "none"
n = max_drones()
Safety. If only main drone exists: it must work
if n <= 1:
return "short"
If we have more drones than dimension, either mode will work.
"long" somehow seems safer. I can't explain.
if n > ws:
return "long"
If workforce is small, subtask-ending-overhead shouldn't be large
if n <= 4:
return "short"
Prefer an even partition using all workers (main included)
if ws % n == 0:
return "short"
If not even, see if excluding the main gives an even split
if ws % (n - 1) == 0:
return "long"
If workforce is small vs width, every hand helps
if (n - 1) < ws // 4:
return "short"
Fallback: pick the mode that yields fewer ragged strips (GCD-based)
def _gcd(a, b):
while b:
a, b = b, a % b
return a
g_all = _gcd(ws, n)
g_nomin = _gcd(ws, n - 1)
Larger GCD → cleaner tiling. If tie, prefer delegator for less contention.
if g_all >= g_nomin:
return "short"
else:
return "long"
def recompute_ws():
global ws
ws = get_world_size()
def recompute_parallelism(new_type=None):
Corner case against runtime upgrades of worldsize or n_drones.
Call at the start of big entry points like till_all() or maintask().
Args:
new_type: manual application of one of "none", "short", "long"
recompute_ws()
global parallelism_type
if new_type == None:
parallelism_type = choose_parallelism(ws)
else:
parallelism_type = new_type
def drone_task(factory, args, local_p_type=None):
This function is for usage with a "N independent jobs" pattern.
Convention:
Ensure subtasks are "closed" w.r.t. starting position. This is
in case of the "short" mode needing main drone to be a worker.
If not, ensure so externally around your function call.
Args:
factory: factory function taking m parameters which
can cleanly create zero-arg subtasks.
args: list of m-tuple of arguments to pass into factory.
(implied) length: number of jobs. Inferred as n=len(args)
local_p_type: local scope parallelism_type.
Useful for second-order delegators.
Returns:
results: list of n results
length = len(args)
if length == 0: # Trivial
return []
if local_p_type != None:
p_type_to_use = local_p_type
else:
global keyword not needed for readonly
p_type_to_use = parallelism_type
_drones = []
for _ in range(length):
_drones.append(None)
results = []
for _ in range(length):
results.append(None)
if p_type_to_use == "none":
Does not account for drone parallelism.
for i in range(length):
res = factory(args[i])()
results[i] = res
return results
elif p_type_to_use == "short":
Uses the main drone as a worker.
for i in range(length):
subtask_watcher = spawn_drone(factory(args[i]))
if subtask_watcher:
_drones[i] = subtask_watcher
else:
res = factory(args[i])()
results[i] = res
for i in range(len(_drones)):
if _drones[i] != None:
results[i] = wait_for(_drones[i])
return results
elif p_type_to_use == "long":