|
1 | 1 | import logging |
| 2 | +from concurrent.futures import ThreadPoolExecutor |
2 | 3 | from datetime import datetime, timedelta, timezone |
3 | 4 | from pathlib import Path |
4 | 5 | from typing import List |
@@ -52,20 +53,40 @@ def sync_scratch_directory( |
52 | 53 | except Exception as e: |
53 | 54 | logger.error("Failed to delete %s: %s", file_path, e) |
54 | 55 |
|
55 | | - # Download missing files |
56 | | - local_kml_paths: List[Path] = [] |
57 | | - for url in urls: |
58 | | - filename = f"{mission_name}_{Path(url).stem}.kml" |
59 | | - file_path = scratch_dir / filename |
| 56 | + # Map each url to its target path, preserving order |
| 57 | + url_paths = [ |
| 58 | + (url, scratch_dir / f"{mission_name}_{Path(url).stem}.kml") for url in urls |
| 59 | + ] |
60 | 60 |
|
61 | | - if file_path.name in missing_files or not file_path.exists(): |
| 61 | + # Determine which files are missing and need downloading |
| 62 | + to_download = [ |
| 63 | + (url, file_path) |
| 64 | + for url, file_path in url_paths |
| 65 | + if file_path.name in missing_files or not file_path.exists() |
| 66 | + ] |
| 67 | + |
| 68 | + # Download missing files concurrently (network-bound) |
| 69 | + failed: set = set() |
| 70 | + if to_download: |
| 71 | + |
| 72 | + def _download(item): |
| 73 | + url, file_path = item |
62 | 74 | try: |
63 | 75 | download_kml(url, str(file_path)) |
| 76 | + return None |
64 | 77 | except Exception as e: |
65 | 78 | logger.error("Failed downloading %s: %s", url, e) |
66 | | - continue |
| 79 | + return file_path |
67 | 80 |
|
68 | | - local_kml_paths.append(file_path) |
| 81 | + with ThreadPoolExecutor(max_workers=min(len(to_download), 8)) as executor: |
| 82 | + for result in executor.map(_download, to_download): |
| 83 | + if result is not None: |
| 84 | + failed.add(result) |
| 85 | + |
| 86 | + # Return local paths in original url order, skipping failed downloads |
| 87 | + local_kml_paths: List[Path] = [ |
| 88 | + file_path for _, file_path in url_paths if file_path not in failed |
| 89 | + ] |
69 | 90 |
|
70 | 91 | return local_kml_paths |
71 | 92 |
|
@@ -101,51 +122,63 @@ def build_sentinel_collection( |
101 | 122 | if platforms: |
102 | 123 | platform_by_name = {Path(u).stem.lower(): p for u, p in zip(urls, platforms)} |
103 | 124 |
|
104 | | - gdfs: list[gpd.GeoDataFrame] = [] |
105 | | - |
106 | | - for kml_path in local_kml_paths: |
| 125 | + def _resolve_platform(kml_path: Path) -> str | None: |
| 126 | + if not platform_by_name: |
| 127 | + return None |
| 128 | + stem = kml_path.stem.lower() |
| 129 | + # first attempt: direct match |
| 130 | + platform = platform_by_name.get(stem) |
| 131 | + # second attempt: drop leading token |
| 132 | + if platform is None and "_" in stem: |
| 133 | + stem_id = "_".join(stem.split("_")[1:]) |
| 134 | + platform = platform_by_name.get(stem_id) |
| 135 | + # last resort: partial match |
| 136 | + if platform is None: |
| 137 | + for key, value in platform_by_name.items(): |
| 138 | + if key in stem: |
| 139 | + platform = value |
| 140 | + break |
| 141 | + return platform |
| 142 | + |
| 143 | + def _load_kml(kml_path: Path) -> gpd.GeoDataFrame | None: |
| 144 | + """Read cached geojson or parse KML (CPU-bound), tag with platform.""" |
107 | 145 | collection_path = SCRATCH_DIR / f"{kml_path.stem}.geojson" |
108 | | - platform = None |
109 | | - |
110 | | - if platform_by_name: |
111 | | - stem = kml_path.stem.lower() |
112 | | - # first attempt: direct match |
113 | | - platform = platform_by_name.get(stem) |
114 | | - |
115 | | - # second attempt: drop leading token |
116 | | - if platform is None and "_" in stem: |
117 | | - stem_id = "_".join(stem.split("_")[1:]) |
118 | | - platform = platform_by_name.get(stem_id) |
119 | | - |
120 | | - # last resort: partial match |
121 | | - if platform is None: |
122 | | - for key, value in platform_by_name.items(): |
123 | | - if key in stem: |
124 | | - platform = value |
125 | | - break |
126 | 146 |
|
127 | 147 | if collection_path.exists(): |
128 | | - logger.info("Using cached file: %s", collection_path) |
| 148 | + logger.debug("Using cached file: %s", collection_path) |
129 | 149 | try: |
130 | 150 | gdf = gpd.read_file(collection_path) |
131 | 151 | except Exception as e: |
132 | 152 | logger.error("Failed reading %s: %s", collection_path, e) |
133 | | - continue |
| 153 | + return None |
134 | 154 | else: |
135 | | - logger.info("Parsing new file: %s", kml_path) |
| 155 | + logger.debug("Parsing new file: %s", kml_path) |
136 | 156 | try: |
137 | 157 | gdf = parse_kml(kml_path) |
138 | 158 | if not gdf.empty: |
139 | 159 | gdf.to_file(collection_path) |
140 | 160 | else: |
141 | 161 | logger.warning("No valid data in file: %s", kml_path) |
142 | | - continue |
| 162 | + return None |
143 | 163 | except Exception as e: |
144 | 164 | logger.error("Failed parsing %s: %s", kml_path, e) |
145 | | - continue |
146 | | - |
147 | | - gdf["platform"] = platform |
148 | | - gdfs.append(gdf) |
| 165 | + return None |
| 166 | + |
| 167 | + gdf["platform"] = _resolve_platform(kml_path) |
| 168 | + return gdf |
| 169 | + |
| 170 | + # Parse/read each KML concurrently. Order is irrelevant: the results are |
| 171 | + # concatenated and re-sorted by begin_date below. Each writes a distinct |
| 172 | + # geojson path, so there is no write collision. |
| 173 | + if local_kml_paths: |
| 174 | + with ThreadPoolExecutor(max_workers=min(len(local_kml_paths), 8)) as executor: |
| 175 | + gdfs = [ |
| 176 | + gdf |
| 177 | + for gdf in executor.map(_load_kml, local_kml_paths) |
| 178 | + if gdf is not None |
| 179 | + ] |
| 180 | + else: |
| 181 | + gdfs = [] |
149 | 182 |
|
150 | 183 | if not gdfs: |
151 | 184 | logger.error("No valid GeoDataFrames created.") |
|
0 commit comments