Spaces:
Sleeping
Sleeping
| import numpy as np | |
| import random | |
| import math | |
| class _Instance: | |
| """A multi-cell rectangular item placed in the truck lattice.""" | |
| __slots__ = ("id", "client", "item_type", "is_return", "anchor", "shape") | |
| def __init__(self, instance_id, client, item_type, is_return, anchor, shape): | |
| self.id = instance_id | |
| self.client = client | |
| self.item_type = item_type | |
| self.is_return = is_return | |
| self.anchor = anchor # (x, y, z) min corner | |
| self.shape = shape # (lx, ly, lz) | |
| def cells_at(self, anchor): | |
| x0, y0, z0 = anchor | |
| lx, ly, lz = self.shape | |
| out = [] | |
| for i in range(lx): | |
| for j in range(ly): | |
| for k in range(lz): | |
| out.append((x0 + i, y0 + j, z0 + k)) | |
| return out | |
| def cells(self): | |
| return self.cells_at(self.anchor) | |
| class SmartTruckOptimizer3D: | |
| EMPTY_KEG = -1 # post-return solid block; structural support but always blocks extraction | |
| def __init__(self, length_bays, width_pallets, height_layers, route, item_shapes=None): | |
| """ | |
| length_bays: lateral bays (X-axis) | |
| width_pallets: depth of each bay (Y-axis); both Y-walls are open along the full length | |
| height_layers: max stacking height (Z-axis) | |
| route: client IDs in delivery order | |
| item_shapes: dict[item_type, (lx, ly, lz)] of rectangular box dimensions. | |
| Defaults to {"keg": (1, 1, 1)} for backward compatibility. | |
| Removability rule: an instance is removable iff at least one of its three sides | |
| — top, left (-Y), right (+Y) — is fully clear of blockers (other-client items | |
| or empty kegs). The truck has no front/back access. | |
| """ | |
| self.L = length_bays | |
| self.W = width_pallets | |
| self.H = height_layers | |
| self.route = list(route) | |
| shapes = item_shapes if item_shapes is not None else {"keg": (1, 1, 1)} | |
| self.item_shapes = {} | |
| for name, dims in shapes.items(): | |
| dims = tuple(int(d) for d in dims) | |
| if len(dims) != 3 or any(d < 1 for d in dims): | |
| raise ValueError(f"Invalid shape for item '{name}': {dims}") | |
| if dims[0] > self.L or dims[1] > self.W or dims[2] > self.H: | |
| raise ValueError(f"Item '{name}' shape {dims} exceeds grid {(self.L, self.W, self.H)}.") | |
| self.item_shapes[name] = dims | |
| self._instances = {} | |
| self._next_id = 1 | |
| # Work caches (variance over cell-x; multi-cell items contribute each cell once) | |
| self._x_sum = {} | |
| self._x_sq_sum = {} | |
| self._counts = {} | |
| self._max_var = max(((self.L - 1) ** 2) / 4.0, 1e-9) | |
| # ---------- helpers ---------- | |
| def _is_blocker(self, value, client): | |
| """True iff a cell with `value` blocks `client` from extracting through it.""" | |
| if value == 0: | |
| return False | |
| if value == self.EMPTY_KEG: | |
| return True | |
| return self._instances[value].client != client | |
| # ---------- initial state ---------- | |
| def generate_initial_state(self, client_item_counts, client_returns): | |
| """ | |
| client_item_counts: {client_id: {item_type: total_count}} | |
| client_returns: {client_id: {item_type: returns_count}} | |
| For each (client, item_type) we create `total - returns` pure-delivery instances | |
| and `returns` substitution instances. Instances are placed at random valid | |
| anchors; raises if the grid is too crowded. | |
| """ | |
| self._instances = {} | |
| self._next_id = 1 | |
| state = np.zeros((self.L, self.W, self.H), dtype=int) | |
| pending = [] # (client, item_type, is_return, shape) | |
| for client, items in client_item_counts.items(): | |
| returns_for_client = client_returns.get(client, {}) | |
| for item_type, count in items.items(): | |
| if item_type not in self.item_shapes: | |
| raise KeyError(f"Unknown item_type '{item_type}' (not in item_shapes).") | |
| num_returns = returns_for_client.get(item_type, 0) | |
| num_pure = count - num_returns | |
| if num_pure < 0: | |
| raise ValueError( | |
| f"Client {client} item '{item_type}': returns ({num_returns}) " | |
| f"exceed total ({count})." | |
| ) | |
| shape = self.item_shapes[item_type] | |
| for _ in range(num_pure): | |
| pending.append((client, item_type, False, shape)) | |
| for _ in range(num_returns): | |
| pending.append((client, item_type, True, shape)) | |
| total_cells_needed = sum(s[0] * s[1] * s[2] for *_, s in pending) | |
| capacity = self.L * self.W * self.H | |
| if total_cells_needed > capacity: | |
| raise ValueError(f"Items need {total_cells_needed} cells; grid holds {capacity}.") | |
| random.shuffle(pending) | |
| for client, item_type, is_return, shape in pending: | |
| anchor = self._place_random(state, shape) | |
| if anchor is None: | |
| raise RuntimeError( | |
| f"Could not place item '{item_type}' for client {client}; grid too crowded." | |
| ) | |
| inst_id = self._next_id | |
| self._next_id += 1 | |
| inst = _Instance(inst_id, client, item_type, is_return, anchor, shape) | |
| self._instances[inst_id] = inst | |
| for (x, y, z) in inst.cells: | |
| state[x, y, z] = inst_id | |
| return state | |
| def _place_random(self, state, shape, max_tries=500): | |
| lx, ly, lz = shape | |
| for _ in range(max_tries): | |
| x = random.randint(0, self.L - lx) | |
| y = random.randint(0, self.W - ly) | |
| z = random.randint(0, self.H - lz) | |
| if self._cells_clear(state, x, y, z, lx, ly, lz): | |
| return (x, y, z) | |
| # Deterministic fallback scan | |
| for x in range(self.L - lx + 1): | |
| for y in range(self.W - ly + 1): | |
| for z in range(self.H - lz + 1): | |
| if self._cells_clear(state, x, y, z, lx, ly, lz): | |
| return (x, y, z) | |
| return None | |
| def _cells_clear(self, state, x, y, z, lx, ly, lz, allow_id=0): | |
| """True iff the box at (x,y,z) of size (lx,ly,lz) contains only 0 or `allow_id`.""" | |
| for i in range(lx): | |
| for j in range(ly): | |
| for k in range(lz): | |
| v = state[x + i, y + j, z + k] | |
| if v != 0 and v != allow_id: | |
| return False | |
| return True | |
| # ---------- penalty (full eval) ---------- | |
| def physical_penalty(self, state): | |
| """ | |
| Sum of: | |
| 1. Initial gravity violations (cells whose support is void) | |
| 2. For each client in route order: | |
| a. Extraction violations: instances with no fully-clear side | |
| b. Post-ablation gravity violations | |
| """ | |
| penalty = 0 | |
| truck = np.copy(state) | |
| penalty += self._gravity_violations(truck) | |
| # Group instances by client once | |
| by_client = {c: [] for c in self.route} | |
| for inst in self._instances.values(): | |
| if inst.client in by_client: | |
| by_client[inst.client].append(inst) | |
| for client in self.route: | |
| for inst in by_client[client]: | |
| if self._instance_blocked(truck, inst): | |
| penalty += 1 | |
| for inst in by_client[client]: | |
| fill = self.EMPTY_KEG if inst.is_return else 0 | |
| for (x, y, z) in inst.cells: | |
| truck[x, y, z] = fill | |
| penalty += self._gravity_violations(truck) | |
| return penalty | |
| def _gravity_violations(self, truck): | |
| if self.H < 2: | |
| return 0 | |
| upper = truck[:, :, 1:] | |
| lower = truck[:, :, :-1] | |
| return int(np.sum((upper > 0) & (lower == 0))) | |
| def _instance_blocked(self, truck, inst): | |
| """True iff none of {top, left, right} sides is fully clear of blockers.""" | |
| x0, y0, z0 = inst.anchor | |
| lx, ly, lz = inst.shape | |
| c = inst.client | |
| # Top side: above the item | |
| if z0 + lz >= self.H: | |
| return False # extends to roof, top is open by definition | |
| if self._region_clear(truck, x0, x0 + lx, y0, y0 + ly, z0 + lz, self.H, c): | |
| return False | |
| # Left side | |
| if y0 == 0: | |
| return False | |
| if self._region_clear(truck, x0, x0 + lx, 0, y0, z0, z0 + lz, c): | |
| return False | |
| # Right side | |
| if y0 + ly >= self.W: | |
| return False | |
| if self._region_clear(truck, x0, x0 + lx, y0 + ly, self.W, z0, z0 + lz, c): | |
| return False | |
| return True | |
| def _region_clear(self, truck, x_lo, x_hi, y_lo, y_hi, z_lo, z_hi, client): | |
| for x in range(x_lo, x_hi): | |
| for y in range(y_lo, y_hi): | |
| for z in range(z_lo, z_hi): | |
| if self._is_blocker(truck[x, y, z], client): | |
| return False | |
| return True | |
| # ---------- delta penalty ---------- | |
| def _instances_reading_cells(self, changed_cells): | |
| """Set of instance IDs whose blocking computation reads any of `changed_cells`.""" | |
| affected = set() | |
| for inst_id, inst in self._instances.items(): | |
| x0, y0, z0 = inst.anchor | |
| lx, ly, lz = inst.shape | |
| for (cx, cy, cz) in changed_cells: | |
| in_x = x0 <= cx < x0 + lx | |
| in_y = y0 <= cy < y0 + ly | |
| in_z = z0 <= cz < z0 + lz | |
| if in_x and ( | |
| (in_y and cz >= z0 + lz) or # top column | |
| (cy < y0 and in_z) or # left row | |
| (cy >= y0 + ly and in_z) or # right row | |
| (in_y and in_z) # inside (shouldn't happen unless self) | |
| ): | |
| affected.add(inst_id) | |
| break | |
| return affected | |
| def _evaluate_local_penalty(self, state, changed_cells, affected_ids): | |
| """ | |
| Sum penalty contributions whose values can differ between current and proposed | |
| states. Cells outside `changed_cells` and instances outside `affected_ids` give | |
| identical contributions in both, so they cancel in the delta. | |
| The cascade still mutates a full-state copy (ablation is global), but only the | |
| listed instances' extraction blocks and only the gravity neighbors of changed | |
| cells contribute to the returned sum. | |
| """ | |
| truck = np.copy(state) | |
| gravity_cells = set() | |
| for (x, y, z) in changed_cells: | |
| gravity_cells.add((x, y, z)) | |
| if z + 1 < self.H: | |
| gravity_cells.add((x, y, z + 1)) | |
| penalty = 0 | |
| for (x, y, z) in gravity_cells: | |
| if z >= 1 and truck[x, y, z] > 0 and truck[x, y, z - 1] == 0: | |
| penalty += 1 | |
| by_client = {c: [] for c in self.route} | |
| for inst in self._instances.values(): | |
| if inst.client in by_client: | |
| by_client[inst.client].append(inst) | |
| for client in self.route: | |
| for inst in by_client[client]: | |
| if inst.id in affected_ids and self._instance_blocked(truck, inst): | |
| penalty += 1 | |
| for inst in by_client[client]: | |
| fill = self.EMPTY_KEG if inst.is_return else 0 | |
| for (x, y, z) in inst.cells: | |
| truck[x, y, z] = fill | |
| for (x, y, z) in gravity_cells: | |
| if z >= 1 and truck[x, y, z] > 0 and truck[x, y, z - 1] == 0: | |
| penalty += 1 | |
| return penalty | |
| # ---------- work ---------- | |
| def spatial_work(self, state): | |
| """ | |
| Mean-normalized variance of x-coords per client, in roughly [0, 1]. | |
| Each cell of a multi-cell item contributes its x once; rewards keeping a | |
| client's items clustered along the truck length so the driver walks less | |
| per stop. Translation-invariant (cluster *position* doesn't matter — both | |
| Y-sides are open, see project notes). | |
| """ | |
| if not self.route: | |
| return 0.0 | |
| total = 0.0 | |
| for client in self.route: | |
| xs = [] | |
| for inst in self._instances.values(): | |
| if inst.client == client: | |
| x0, _, _ = inst.anchor | |
| lx, ly, lz = inst.shape | |
| for i in range(lx): | |
| xs.extend([x0 + i] * (ly * lz)) | |
| if len(xs) > 1: | |
| total += float(np.var(xs)) | |
| return total / (self._max_var * len(self.route)) | |
| def _init_work_cache(self, state): | |
| self._x_sum = {c: 0 for c in self.route} | |
| self._x_sq_sum = {c: 0 for c in self.route} | |
| self._counts = {c: 0 for c in self.route} | |
| for inst in self._instances.values(): | |
| if inst.client not in self._counts: | |
| continue | |
| for (x, _, _) in inst.cells: | |
| self._x_sum[inst.client] += x | |
| self._x_sq_sum[inst.client] += x * x | |
| self._counts[inst.client] += 1 | |
| def _work_from_cache(self): | |
| if not self.route: | |
| return 0.0 | |
| total = 0.0 | |
| for client in self.route: | |
| n = self._counts.get(client, 0) | |
| if n > 1: | |
| mean = self._x_sum[client] / n | |
| total += self._x_sq_sum[client] / n - mean * mean | |
| return total / (self._max_var * len(self.route)) | |
| def _update_work_cache_for_move(self, inst, old_anchor, new_anchor): | |
| """Reflect that `inst` translated from old_anchor to new_anchor. O(|cells|).""" | |
| if inst.client not in self._counts: | |
| return | |
| for (x, _, _) in inst.cells_at(old_anchor): | |
| self._x_sum[inst.client] -= x | |
| self._x_sq_sum[inst.client] -= x * x | |
| for (x, _, _) in inst.cells_at(new_anchor): | |
| self._x_sum[inst.client] += x | |
| self._x_sq_sum[inst.client] += x * x | |
| # counts unchanged | |
| # ---------- optimize ---------- | |
| def optimize(self, initial_state, steps=40000, seed=None, | |
| T_0=1.0, T_min=1e-4, gamma_0=0.01, gamma_max=100.0): | |
| """ | |
| SA with single-instance translation proposals + delta penalty evaluation. | |
| Each step picks a random instance, picks a random in-bounds anchor, and | |
| accepts/rejects a translation by Metropolis criterion on H = W + γP. Same-anchor | |
| proposals and proposals that overlap other items are skipped without scoring. | |
| """ | |
| if seed is not None: | |
| random.seed(seed) | |
| np.random.seed(seed) | |
| current_state = np.copy(initial_state) | |
| self._init_work_cache(current_state) | |
| current_P = self.physical_penalty(current_state) | |
| current_W = self._work_from_cache() | |
| current_H = current_W + gamma_max * current_P | |
| best_state = np.copy(current_state) | |
| best_H = current_H | |
| best_feasible_state = None | |
| best_feasible_W = float('inf') | |
| instance_ids = list(self._instances.keys()) | |
| if not instance_ids: | |
| return current_state, current_P, current_W | |
| for step in range(steps): | |
| fraction = step / float(steps) | |
| T = T_0 * ((T_min / T_0) ** fraction) | |
| gamma = gamma_0 + (gamma_max - gamma_0) * (fraction ** 4) | |
| inst_id = random.choice(instance_ids) | |
| inst = self._instances[inst_id] | |
| lx, ly, lz = inst.shape | |
| x_new = random.randint(0, self.L - lx) | |
| y_new = random.randint(0, self.W - ly) | |
| z_new = random.randint(0, self.H - lz) | |
| new_anchor = (x_new, y_new, z_new) | |
| old_anchor = inst.anchor | |
| if new_anchor == old_anchor: | |
| continue | |
| # Validate: new footprint is void or self-owned | |
| if not self._cells_clear(current_state, x_new, y_new, z_new, lx, ly, lz, allow_id=inst_id): | |
| continue | |
| old_cells = inst.cells_at(old_anchor) | |
| new_cells = inst.cells_at(new_anchor) | |
| changed_cells = set(old_cells) | set(new_cells) | |
| affected_ids = self._instances_reading_cells(changed_cells) | |
| affected_ids.add(inst_id) | |
| # Score before | |
| before_local = self._evaluate_local_penalty(current_state, changed_cells, affected_ids) | |
| # Apply move | |
| for (x, y, z) in old_cells: | |
| current_state[x, y, z] = 0 | |
| for (x, y, z) in new_cells: | |
| current_state[x, y, z] = inst_id | |
| inst.anchor = new_anchor | |
| self._update_work_cache_for_move(inst, old_anchor, new_anchor) | |
| # Score after (note: instance.anchor is now new_anchor, so its blocking is | |
| # evaluated at the new position; affected_ids set doesn't change because it | |
| # already included this instance.) | |
| after_local = self._evaluate_local_penalty(current_state, changed_cells, affected_ids) | |
| delta_P = after_local - before_local | |
| proposed_P = current_P + delta_P | |
| proposed_W = self._work_from_cache() | |
| delta_W = proposed_W - current_W | |
| delta_H = delta_W + gamma * delta_P | |
| if delta_H < 0 or random.random() < math.exp(-delta_H / max(T, 1e-12)): | |
| current_P = proposed_P | |
| current_W = proposed_W | |
| current_H = current_W + gamma_max * current_P | |
| if current_H < best_H: | |
| best_H = current_H | |
| best_state = np.copy(current_state) | |
| if current_P == 0 and current_W < best_feasible_W: | |
| best_feasible_W = current_W | |
| best_feasible_state = np.copy(current_state) | |
| else: | |
| # Revert | |
| for (x, y, z) in new_cells: | |
| current_state[x, y, z] = 0 | |
| for (x, y, z) in old_cells: | |
| current_state[x, y, z] = inst_id | |
| inst.anchor = old_anchor | |
| self._update_work_cache_for_move(inst, new_anchor, old_anchor) | |
| result_state = best_feasible_state if best_feasible_state is not None else best_state | |
| # Sync instance anchors to the returned state — best_state was np.copy'd at peak | |
| # quality, but self._instances has been moving along with current_state since. | |
| self._sync_instances_to_state(result_state) | |
| return result_state, self.physical_penalty(result_state), self.spatial_work(result_state) | |
| # ---------- result inspection ---------- | |
| def _sync_instances_to_state(self, state): | |
| """Set each instance's anchor to its min-corner in `state`. Drops instances | |
| whose cells aren't found (shouldn't happen for an internally produced state).""" | |
| missing = [] | |
| for inst_id, inst in self._instances.items(): | |
| cells = np.argwhere(state == inst_id) | |
| if len(cells) == 0: | |
| missing.append(inst_id) | |
| continue | |
| mn = cells.min(axis=0) | |
| inst.anchor = (int(mn[0]), int(mn[1]), int(mn[2])) | |
| for inst_id in missing: | |
| del self._instances[inst_id] | |
| def get_layout(self, state=None): | |
| """ | |
| Return the optimized placement as a list of dicts, one per instance: | |
| {id, client, item_type, is_return, anchor, shape, cells} | |
| With no argument, reads the current synced anchors (set by optimize()). | |
| Pass a state array to derive positions directly from it instead. | |
| """ | |
| layout = [] | |
| if state is None: | |
| for inst in self._instances.values(): | |
| layout.append({ | |
| "id": inst.id, | |
| "client": inst.client, | |
| "item_type": inst.item_type, | |
| "is_return": inst.is_return, | |
| "anchor": inst.anchor, | |
| "shape": inst.shape, | |
| "cells": inst.cells, | |
| }) | |
| else: | |
| for inst in self._instances.values(): | |
| cells = np.argwhere(state == inst.id) | |
| if len(cells) == 0: | |
| continue | |
| anchor = tuple(int(v) for v in cells.min(axis=0)) | |
| layout.append({ | |
| "id": inst.id, | |
| "client": inst.client, | |
| "item_type": inst.item_type, | |
| "is_return": inst.is_return, | |
| "anchor": anchor, | |
| "shape": inst.shape, | |
| "cells": [tuple(int(v) for v in c) for c in cells], | |
| }) | |
| return layout | |
| # --- EXECUTION --- | |
| if __name__ == "__main__": | |
| optimizer = SmartTruckOptimizer3D( | |
| length_bays=8, width_pallets=4, height_layers=3, route=[1, 2, 3], | |
| item_shapes={ | |
| "keg": (1, 1, 1), | |
| "crate": (2, 1, 1), | |
| "tower": (1, 1, 2), | |
| }, | |
| ) | |
| # {client: {item_type: total_count}} | |
| orders = { | |
| 1: {"keg": 12, "crate": 4}, | |
| 2: {"keg": 18, "crate": 6}, | |
| 3: {"keg": 15, "tower": 4}, | |
| } | |
| # {client: {item_type: returns_count}} — must be ≤ total per (client, item) | |
| returns = { | |
| 1: {"keg": 8}, | |
| 2: {"keg": 5}, | |
| 3: {"keg": 10}, | |
| } | |
| initial = optimizer.generate_initial_state(orders, returns) | |
| print("Initial 3D Penalty: ", optimizer.physical_penalty(initial)) | |
| print(f"Initial Work (normalized): {optimizer.spatial_work(initial):.4f}") | |
| final_state, final_P, final_W = optimizer.optimize(initial, steps=30000) | |
| print(f"\nOptimization Complete.") | |
| print(f"Final Physical Violations: {final_P}") | |
| print(f"Final Work Metric: {final_W:.4f}") | |
| if final_P == 0: | |
| print("\nValid 3D Lattice Layout generated successfully.") | |
| else: | |
| print("\nOptimizer trapped in local minimum. Try increasing 'steps'.") | |
| layout = optimizer.get_layout() | |
| print(f"\nLayout: {len(layout)} items") | |
| for item in sorted(layout, key=lambda i: (i["client"], i["anchor"])): | |
| kind = "return" if item["is_return"] else "delivery" | |
| print(f" client={item['client']} {item['item_type']:<6} {kind:<8} " | |
| f"anchor={item['anchor']} shape={item['shape']}") | |
| ''' | |
| Input format for generate_initial_state: | |
| orders = {client_id: {item_type: total_count}} | |
| returns = {client_id: {item_type: returns_count}} | |
| ''' |