Apply sign work zones before candidate routing

mosfet/world.py

@@ -76,6 +76,113 @@ class World:
         result.sort(key=lambda sign: utils.phyp(center, sign['position']))
         return result
 
+    def sign_policies(self, bot_name, distance=64):
+        """Parse sign policies addressed to this bot or its numbered group."""
+        bot_name = str(bot_name).casefold()
+        group_name = (bot_name[:-1] if len(bot_name) > 1 and
+                      bot_name[-1] in '123' else None)
+        addresses = {bot_name}
+        if group_name:
+            addresses.add(group_name)
+
+        center = utils.pint(self.g.pos)
+        policies = []
+        active_sign_position = None
+        for sign in self.find_signs(center, distance):
+            lines = [line.strip() for line in sign['lines'] if line.strip()]
+            if not lines:
+                continue
+            first = lines[0].split(maxsplit=1)
+            address = first[0].rstrip(':,').casefold()
+            if address not in addresses:
+                continue
+            directives = ([first[1]] if len(first) > 1 else []) + lines[1:]
+            text = ' '.join(directives).casefold()
+            sign_policies = []
+            for match in re.finditer(
+                    r'\bstay\s+within\s*(\d+)(?:\s+(squares?|circle))?\b',
+                    text):
+                shape = match.group(2) or 'circle'
+                if shape == 'squares':
+                    shape = 'square'
+                sign_policies.append({
+                    'position': sign['position'],
+                    'distance': int(match.group(1)),
+                    'shape': shape,
+                    'side': sign['side'],
+                })
+            if not sign_policies:
+                continue
+            if active_sign_position is None:
+                # The nearest addressed policy sign overrides farther signs.
+                active_sign_position = sign['position']
+            if sign['position'] == active_sign_position:
+                policies.extend(sign_policies)
+        return policies
+
+    def _active_work_policies(self):
+        job = self.g.get('job')
+        machine = getattr(job, 'active_machine', None) if job else None
+        if machine is None:
+            return []
+        support_machines = (
+            'grab_supplies_states', 'sleep_with_bed_states',
+            'eat_food_states', 'cache_items_states',
+            'check_threats_states',
+        )
+        if any(machine is getattr(job, name, None)
+               for name in support_machines):
+            return []
+        return self.sign_policies(self.g.get('name', ''))
+
+    @staticmethod
+    def _sign_policy_distance(position, policy):
+        # Work zones are horizontal; crop height or terrain elevation does not
+        # consume the sign's radius.
+        dx = abs(position[0] - policy['position'][0])
+        dz = abs(position[2] - policy['position'][2])
+        if policy['shape'] == 'square':
+            return max(dx, dz)
+        return utils.hypot(dx, dz)
+
+    @classmethod
+    def _inside_sign_policy(cls, position, policy):
+        return cls._sign_policy_distance(position, policy) <= policy['distance']
+
+    def work_position_allowed(self, position):
+        """Whether a work target satisfies all currently active sign rules."""
+        return all(self._inside_sign_policy(position, policy)
+                   for policy in self._active_work_policies())
+
+    def filter_work_positions(self, positions):
+        """Filter a candidate set once against the active sign policies."""
+        policies = self._active_work_policies()
+        positions = list(positions)
+        if not policies:
+            return positions
+        return [position for position in positions
+                if all(self._inside_sign_policy(position, policy)
+                       for policy in policies)]
+
+    @classmethod
+    def _path_allowed_by_sign(cls, points, policy):
+        distances = [cls._sign_policy_distance(point, policy)
+                     for point in points]
+        radius = policy['distance']
+        if distances[0] <= radius:
+            return all(distance <= radius for distance in distances)
+
+        # If the bot starts just outside a newly read boundary, let it return
+        # without allowing it to travel farther away from that sign.
+        start_distance = distances[0]
+        entered = False
+        for distance in distances[1:]:
+            if distance > start_distance or (entered and distance > radius):
+                return False
+            if distance <= radius:
+                entered = True
+        return entered
+
     def check_air_column(self, pos, distance):
         for i in range(distance):
             check = utils.padd(pos, (0, i, 0))
@@ -194,11 +301,23 @@ class World:
         maze_solver = path.Pathfinder(self.g)
 
         try:
-            s = maze_solver.astar(start, place)
-            return list(s) if s else None
+            solution = maze_solver.astar(start, place)
+            if not solution:
+                return None
+            navpath = list(solution)
         except path.AStarTimeout:
             return None
 
+        policies = self._active_work_policies()
+        points = [tuple(start)] + [tuple(point) for point in navpath]
+        for policy in policies:
+            if not self._path_allowed_by_sign(points, policy):
+                print('[sign policy] rejected work path outside {}-block {} '
+                      'around {}'.format(policy['distance'], policy['shape'],
+                                         policy['position']), flush=True)
+                return None
+        return navpath
+
     def path_to_place_faked(self, start, place):
         # same as above, but adds a fake block below and air before pathfinding
         # so that the pathfinder can actually make it to the block