Spaces:
Running
Running
| import hashlib | |
| import json | |
| import os | |
| from typing import Dict, Tuple, List, Optional | |
| class SpatialGridManager: | |
| """ | |
| Manages a 2D grid representation of the Agent Matrix. | |
| Grid Size: 9 rows x 8 columns = 72 Nodes. | |
| """ | |
| GRID_ROWS = 9 | |
| GRID_COLS = 8 | |
| def __init__(self, index_path: str = "workspace/spatial_index.json"): | |
| self.index_path = index_path | |
| self.agent_to_coords: Dict[str, Tuple[int, int]] = {} | |
| self.coords_to_agent: Dict[Tuple[int, int], str] = {} | |
| self._initialize_grid() | |
| def _initialize_grid(self): | |
| # Load the agent keywords from the matrix JSON | |
| matrix_path = os.path.join(os.path.dirname(__file__), "..", "agents", "matrix_agents.json") | |
| if not os.path.exists(matrix_path): | |
| return | |
| with open(matrix_path, "r") as f: | |
| agents = json.load(f) | |
| # Map each agent to a coordinate linearly | |
| for i, agent in enumerate(agents): | |
| if i >= self.GRID_ROWS * self.GRID_COLS: | |
| break | |
| x, y = divmod(i, self.GRID_COLS) | |
| keyword = agent["keyword"] | |
| self.agent_to_coords[keyword] = (x, y) | |
| self.coords_to_agent[(x, y)] = keyword | |
| def map_uri_to_coord(self, uri: str) -> Tuple[int, int]: | |
| """ | |
| URI-based spatial indexing. | |
| Hashes the URI to determine its 'spatial' coordinate in the matrix. | |
| """ | |
| hash_val = int(hashlib.sha256(uri.encode()).hexdigest(), 16) | |
| index = hash_val % (self.GRID_ROWS * self.GRID_COLS) | |
| return divmod(index, self.GRID_COLS) | |
| def sync_data_to_node(self, uri: str, content: dict): | |
| """Automated data synchronization to the grid node mapped by the URI.""" | |
| x, y = self.map_uri_to_coord(uri) | |
| agent_key = self.coords_to_agent.get((x, y)) | |
| if not agent_key: | |
| return | |
| # Load or initialize the spatial index | |
| index_data = {} | |
| if os.path.exists(self.index_path): | |
| with open(self.index_path, "r") as f: | |
| index_data = json.load(f) | |
| # Update node data | |
| if agent_key not in index_data: | |
| index_data[agent_key] = [] | |
| # Keep only latest updates (sliding window sync) | |
| index_data[agent_key].append({ | |
| "uri": uri, | |
| "data": content, | |
| "timestamp": content.get("published", "") | |
| }) | |
| index_data[agent_key] = index_data[agent_key][-5:] # Limit to last 5 items per node | |
| os.makedirs(os.path.dirname(self.index_path), exist_ok=True) | |
| with open(self.index_path, "w") as f: | |
| json.dump(index_data, f, indent=2) | |
| return agent_key, (x, y) | |
| def get_node_data(self, agent_keyword: str) -> List[dict]: | |
| if not os.path.exists(self.index_path): | |
| return [] | |
| with open(self.index_path, "r") as f: | |
| index_data = json.load(f) | |
| return index_data.get(agent_keyword, []) | |