Add new items to the work queue (local version). Args: work_paths: Each individual path (local in this context) that we will process over items_per_group: Number of items to group together in a single work item
(self, work_paths: List[str], items_per_group: int)
| 193 | self._queue: Queue[Any] = Queue() |
| 194 | |
| 195 | async def populate_queue(self, work_paths: List[str], items_per_group: int) -> None: |
| 196 | """ |
| 197 | Add new items to the work queue (local version). |
| 198 | |
| 199 | Args: |
| 200 | work_paths: Each individual path (local in this context) |
| 201 | that we will process over |
| 202 | items_per_group: Number of items to group together in a single work item |
| 203 | """ |
| 204 | # Treat them as local paths, but keep variable name for consistency |
| 205 | all_paths = set(work_paths) |
| 206 | logger.info(f"Found {len(all_paths):,} total paths") |
| 207 | |
| 208 | # Load existing work groups from local index |
| 209 | existing_lines = await asyncio.to_thread(download_zstd_csv_local, self._index_path) |
| 210 | existing_groups = {} |
| 211 | for line in existing_lines: |
| 212 | if line.strip(): |
| 213 | parts = line.strip().split(",") |
| 214 | group_hash = parts[0] |
| 215 | group_paths = parts[1:] |
| 216 | existing_groups[group_hash] = group_paths |
| 217 | |
| 218 | existing_path_set = {p for paths in existing_groups.values() for p in paths} |
| 219 | new_paths = all_paths - existing_path_set |
| 220 | logger.info(f"{len(new_paths):,} new paths to add to the workspace") |
| 221 | |
| 222 | if not new_paths: |
| 223 | return |
| 224 | |
| 225 | # Create new work groups |
| 226 | new_groups = [] |
| 227 | current_group = [] |
| 228 | for path in sorted(new_paths): |
| 229 | current_group.append(path) |
| 230 | if len(current_group) == items_per_group: |
| 231 | group_hash = self._compute_workgroup_hash(current_group) |
| 232 | new_groups.append((group_hash, current_group)) |
| 233 | current_group = [] |
| 234 | if current_group: |
| 235 | group_hash = self._compute_workgroup_hash(current_group) |
| 236 | new_groups.append((group_hash, current_group)) |
| 237 | |
| 238 | logger.info(f"Created {len(new_groups):,} new work groups") |
| 239 | |
| 240 | # Combine and save updated work groups |
| 241 | combined_groups = existing_groups.copy() |
| 242 | for group_hash, group_paths in new_groups: |
| 243 | combined_groups[group_hash] = group_paths |
| 244 | |
| 245 | combined_lines = [",".join([group_hash] + group_paths) for group_hash, group_paths in combined_groups.items()] |
| 246 | |
| 247 | if new_groups: |
| 248 | # Write the combined data back to disk in zstd CSV format |
| 249 | await asyncio.to_thread(upload_zstd_csv_local, self._index_path, combined_lines) |
| 250 | |
| 251 | async def initialize_queue(self) -> int: |
| 252 | """ |
no test coverage detected