xopr.stac

22def load_config(
23    config_path: Union[str, Path],
24    overrides: Optional[List[str]] = None,
25    environment: Optional[str] = None
26) -> DictConfig:
27    """
28    Load configuration from YAML file with optional overrides.
29
30    Parameters
31    ----------
32    config_path : Union[str, Path]
33        Path to YAML configuration file
34    overrides : List[str], optional
35        Command-line overrides in dot notation
36        Example: ["data.primary_product=CSARP_qlook", "processing.n_workers=8"]
37    environment : str, optional
38        Environment name to apply (e.g., "production", "test", "development")
39
40    Returns
41    -------
42    DictConfig
43        Configuration object with dot-notation access
44
45    Examples
46    --------
47    >>> conf = load_config("config/catalog.yaml")
48    >>> print(conf.data.primary_product)
49    'CSARP_standard'
50
51    >>> conf = load_config(
52    ...     "config/catalog.yaml",
53    ...     overrides=["processing.n_workers=16"],
54    ...     environment="production"
55    ... )
56    """
57    config_path = Path(config_path)
58    if not config_path.exists():
59        raise FileNotFoundError(f"Configuration file not found: {config_path}")
60
61    # Load user configuration directly
62    conf = OmegaConf.load(config_path)
63
64    # Apply environment-specific overrides if specified
65    if environment and "environments" in conf:
66        if environment in conf.environments:
67            logging.info(f"Applying environment: {environment}")
68            env_conf = conf.environments[environment]
69            conf = OmegaConf.merge(conf, env_conf)
70        else:
71            logging.warning(f"Environment '{environment}' not found in config")
72
73    # Apply command-line overrides
74    if overrides:
75        logging.debug(f"Applying overrides: {overrides}")
76        override_conf = OmegaConf.from_dotlist(overrides)
77        conf = OmegaConf.merge(conf, override_conf)
78
79    # Resolve all interpolations (${...} references)
80    OmegaConf.resolve(conf)
81
82    # Remove environments section from runtime config (no longer needed)
83    if "environments" in conf:
84        del conf["environments"]
85
86    return conf

 89def save_config(conf: DictConfig, output_path: Union[str, Path], add_metadata: bool = True):
 90    """
 91    Save configuration to file for reproducibility.
 92
 93    Parameters
 94    ----------
 95    conf : DictConfig
 96        Configuration to save
 97    output_path : Union[str, Path]
 98        Where to save the configuration
 99    add_metadata : bool
100        Whether to add generation metadata
101    """
102    output_path = Path(output_path)
103    output_path.parent.mkdir(parents=True, exist_ok=True)
104
105    if add_metadata:
106        # Add metadata about when/where this was generated
107        from datetime import datetime
108        save_conf = OmegaConf.create({
109            "_metadata": {
110                "generated_at": datetime.now().isoformat(),
111                "working_directory": os.getcwd(),
112            },
113            **OmegaConf.to_container(conf)
114        })
115    else:
116        save_conf = conf
117
118    OmegaConf.save(save_conf, output_path)
119    logging.info(f"Configuration saved to: {output_path}")

122def validate_config(conf: DictConfig) -> bool:
123    """
124    Basic validation of required configuration fields.
125
126    Parameters
127    ----------
128    conf : DictConfig
129        Configuration to validate
130
131    Returns
132    -------
133    bool
134        True if valid, raises ValueError if not
135    """
136    required_fields = [
137        "data.root",
138        "data.primary_product",
139        "output.path",
140        "output.catalog_id",
141        "output.catalog_description",
142    ]
143
144    for field in required_fields:
145        if OmegaConf.select(conf, field) is None:
146            raise ValueError(f"Required configuration field missing: {field}")
147
148    # Validate n_workers is positive
149    if conf.processing.n_workers <= 0:
150        raise ValueError(f"Invalid n_workers: {conf.processing.n_workers}. Must be positive")
151
152    # Validate paths exist
153    data_root = Path(conf.data.root)
154    if not data_root.exists():
155        raise ValueError(f"Data root does not exist: {data_root}")
156
157    return True

25def create_collection(
26    collection_id: str,
27    description: str,
28    extent: pystac.Extent,
29    license: str = "various",
30    stac_extensions: Optional[List[str]] = None
31) -> pystac.Collection:
32    """
33    Create a STAC collection for a campaign or data product grouping.
34
35    Parameters
36    ----------
37    collection_id : str
38        Unique identifier for the collection.
39    description : str
40        Human-readable description of the collection.
41    extent : pystac.Extent
42        Spatial and temporal extent of the collection.
43    license : str, default ""
44        Data license identifier.
45    stac_extensions : list of str, optional
46        List of STAC extension URLs to enable. If None, defaults to
47        empty list.
48
49    Returns
50    -------
51    pystac.Collection
52        Collection object.
53
54    Examples
55    --------
56    >>> from datetime import datetime
57    >>> import pystac
58    >>> extent = pystac.Extent(
59    ...     spatial=pystac.SpatialExtent([[-180, -90, 180, 90]]),
60    ...     temporal=pystac.TemporalExtent([[datetime(2016, 1, 1), datetime(2016, 12, 31)]])
61    ... )
62    >>> collection = create_collection("2016_campaign", "2016 Antarctic flights", extent)
63    >>> item = create_item("item_001", geometry, bbox, datetime.now())
64    >>> collection.add_item(item)
65    """
66    if stac_extensions is None:
67        stac_extensions = []
68
69    collection = pystac.Collection(
70        id=collection_id,
71        description=description,
72        extent=extent,
73        license=license,
74        stac_extensions=stac_extensions
75    )
76
77    return collection

 80def create_item(
 81    item_id: str,
 82    geometry: Dict[str, Any],
 83    bbox: List[float],
 84    datetime: Any,
 85    properties: Optional[Dict[str, Any]] = None,
 86    assets: Optional[Dict[str, pystac.Asset]] = None,
 87    stac_extensions: Optional[List[str]] = None
 88) -> pystac.Item:
 89    """
 90    Create a STAC item for a flight line data segment.
 91
 92    Parameters
 93    ----------
 94    item_id : str
 95        Unique identifier for the item.
 96    geometry : dict
 97        GeoJSON geometry object.
 98    bbox : list of float
 99        Bounding box coordinates [xmin, ymin, xmax, ymax].
100    datetime : datetime
101        Acquisition datetime.
102    properties : dict, optional
103        Additional metadata properties. If None, defaults to empty dict.
104    assets : dict of str to pystac.Asset, optional
105        Dictionary of assets (data files, thumbnails, etc.). Keys are
106        asset names, values are pystac.Asset objects.
107    stac_extensions : list of str, optional
108        List of STAC extension URLs to enable. If None, defaults to
109        file extension.
110
111    Returns
112    -------
113    pystac.Item
114        Item object with specified properties and assets.
115
116    Examples
117    --------
118    >>> from datetime import datetime
119    >>> import pystac
120    >>> geometry = {"type": "Point", "coordinates": [-71.0, 42.0]}
121    >>> bbox = [-71.1, 41.9, -70.9, 42.1]
122    >>> props = {"instrument": "radar", "platform": "aircraft"}
123    >>> assets = {
124    ...     "data": pystac.Asset(href="https://example.com/data.mat", media_type="application/octet-stream")
125    ... }
126    >>> item = create_item("flight_001", geometry, bbox, datetime.now(), props, assets)
127    """
128    if properties is None:
129        properties = {}
130    if stac_extensions is None:
131        stac_extensions = ['https://stac-extensions.github.io/file/v2.1.0/schema.json']
132
133    item = pystac.Item(
134        id=item_id,
135        geometry=geometry,
136        bbox=bbox,
137        datetime=datetime,
138        properties=properties,
139        stac_extensions=stac_extensions
140    )
141
142    if assets:
143        for key, asset in assets.items():
144            item.add_asset(key, asset)
145
146    return item

149def create_items_from_flight_data(
150    flight_data: Dict[str, Any],
151    config: DictConfig,
152    base_url: str = "https://data.cresis.ku.edu/data/rds/",
153    campaign_name: str = "",
154    primary_data_product: str = "CSARP_standard",
155    provider: str = "cresis",
156    verbose: bool = False,
157    error_log_file: Optional[Union[str, Path]] = None
158) -> List[pystac.Item]:
159    """
160    Create STAC items from flight line data.
161
162    Parameters
163    ----------
164    flight_data : dict
165        Flight metadata from discover_flight_lines(). Expected to contain
166        'flight_id' and 'data_files' keys.
167    config : DictConfig
168        Configuration object with geometry.tolerance setting for simplification.
169    base_url : str, default "https://data.cresis.ku.edu/data/rds/"
170        Base URL for constructing asset hrefs.
171    campaign_name : str, default ""
172        Campaign name for URL construction.
173    primary_data_product : str, default "CSARP_standard"
174        Data product name to use as primary data source.
175    provider : str, default "cresis"
176        Data provider identifier (awi, cresis, dtu, utig).
177    verbose : bool, default False
178        If True, print details for each item being processed.
179    error_log_file : str or Path, optional
180        Path to file where metadata extraction errors will be logged.
181        If None, errors are printed to stdout (default behavior).
182
183    Returns
184    -------
185    list of pystac.Item
186        List of STAC Item objects, one per MAT file in the flight data.
187        Each item contains geometry, temporal information, and asset links.
188    """
189    items = []
190    flight_id = flight_data['flight_id']
191
192    primary_data_files = flight_data['data_files'][primary_data_product].values()
193
194    for data_file_path in primary_data_files:
195        data_path = Path(data_file_path)
196
197        try:
198            # Extract metadata from MAT file only (no CSV needed)
199            metadata = extract_item_metadata(data_path, conf=config)
200        except Exception as e:
201            error_msg = f"Failed to extract metadata for {data_path}: {e}"
202
203            if error_log_file is not None:
204                # Log to file
205                with open(error_log_file, 'a', encoding='utf-8') as f:
206                    f.write(f"{error_msg}\n")
207            else:
208                # Fallback to print (current behavior)
209                print(f"Warning: {error_msg}")
210
211            continue
212
213        item_id = f"{data_path.stem}"
214
215        # Simplify geometry using config tolerance
216        simplified_geom = simplify_geometry_polar_projection(
217            metadata['geom'],
218            simplify_tolerance=config.geometry.tolerance
219        )
220        geometry = mapping(simplified_geom)
221        bbox = list(metadata['bbox'].bounds)
222        datetime = metadata['date']
223
224        # Extract frame number from MAT filename (e.g., "Data_20161014_03_001.mat" -> "001")
225        frame_match = re.search(r'_(\d+)\.mat$', data_path.name)
226        frame = frame_match.group(1)
227
228        # Extract date and segment number from flight_id (e.g., "20161014_03" -> "20161014", "03")
229        # Split on underscore to avoid assuming fixed lengths
230        parts = flight_id.split('_')
231        date_part = parts[0]  # YYYYMMDD
232        segment_num_str = parts[1]  # Segment number as string (formerly flight number)
233
234        # Compute morton bounding box from full (unsimplified) geometry
235        mbox = compute_mbox(mapping(metadata['raw_geom']))
236
237        # Create OPR-specific properties
238        properties = {
239            'opr:provider': provider,
240            'opr:mbox': mbox,
241            'opr:date': date_part,
242            'opr:segment': int(segment_num_str),  # Changed from opr:flight
243            'opr:frame': int(frame)  # Changed from opr:segment
244        }
245
246        # Add scientific extension properties if available
247        item_stac_extensions = [
248            'https://stac-extensions.github.io/file/v2.1.0/schema.json',
249            OPR_EXT,
250        ]
251
252        # Map metadata keys to property names
253        meta_mapping = {
254            'doi': 'sci:doi',
255            'citation': 'sci:citation',
256            'frequency': 'opr:frequency',
257            'bandwidth': 'opr:bandwidth'
258        }
259
260        for key, prop in meta_mapping.items():
261            if metadata.get(key) is not None:
262                value = metadata[key]
263                # Cast frequency/bandwidth to int per OPR extension schema
264                if key in ('frequency', 'bandwidth') and value is not None:
265                    value = int(value)
266                properties[prop] = value
267
268        if any(metadata.get(k) is not None for k in ['doi', 'citation']):
269            item_stac_extensions.append('https://stac-extensions.github.io/scientific/v1.0.0/schema.json')
270
271        assets = {}
272
273        for data_product_type in flight_data['data_files'].keys():
274            if data_path.name in flight_data['data_files'][data_product_type]:
275                product_path = flight_data['data_files'][data_product_type][data_path.name]
276                file_type = metadata.get('mimetype') # get_mat_file_type(product_path)
277                if verbose:
278                    print(f"[{file_type}] {product_path}")
279                assets[data_product_type] = pystac.Asset(
280                    href=base_url + f"{campaign_name}/{data_product_type}/{flight_id}/{data_path.name}",
281                    media_type=file_type
282                )
283                if data_product_type == primary_data_product:
284                    assets['data'] = assets[data_product_type]
285
286        thumb_href = base_url + f"{campaign_name}/images/{flight_id}/{flight_id}_{frame}_2echo_picks.jpg"
287        assets['thumbnail'] = pystac.Asset(
288            href=thumb_href,
289            media_type=pystac.MediaType.JPEG
290        )
291
292        flight_path_href = base_url + f"{campaign_name}/images/{flight_id}/{flight_id}_{frame}_0maps.jpg"
293        assets['flight_path'] = pystac.Asset(
294            href=flight_path_href,
295            media_type=pystac.MediaType.JPEG
296        )
297
298        item = create_item(
299            item_id=item_id,
300            geometry=geometry,
301            bbox=bbox,
302            datetime=datetime,
303            properties=properties,
304            assets=assets,
305            stac_extensions=item_stac_extensions
306        )
307
308        items.append(item)
309
310    return items

377def export_collection_to_parquet(
378    collection: pystac.Collection,
379    config: DictConfig,
380    provider: str = None,
381    hemisphere: str = None
382) -> Optional[Path]:
383    """
384    Export a single STAC collection to a parquet file with collection metadata.
385
386    This function directly converts STAC items to GeoParquet format without
387    intermediate NDJSON, and includes the collection metadata in the Parquet
388    file metadata as per the STAC GeoParquet specification.
389
390    Parameters
391    ----------
392    collection : pystac.Collection
393        STAC collection to export
394    config : DictConfig
395        Configuration object with output.path and logging.verbose settings
396    provider : str, optional
397        Data provider from config (awi, cresis, dtu, utig)
398    hemisphere : str, optional
399        Hemisphere ('north' or 'south'). If not provided, will attempt to detect.
400
401    Returns
402    -------
403    Path or None
404        Path to the created parquet file, or None if no items to export
405
406    Examples
407    --------
408    >>> from omegaconf import OmegaConf
409    >>> config = OmegaConf.create({'output': {'path': './output'}, 'logging': {'verbose': True}})
410    >>> parquet_path = export_collection_to_parquet(collection, config, provider='cresis')
411    >>> print(f"Exported to {parquet_path}")
412    """
413    # Extract settings from config
414    output_dir = Path(config.output.path)
415    verbose = config.logging.get('verbose', False)
416
417    # Get items from collection and subcollections
418    collection_items = list(collection.get_items())
419    if not collection_items:
420        for child_collection in collection.get_collections():
421            collection_items.extend(list(child_collection.get_items()))
422
423    if not collection_items:
424        if verbose:
425            print(f"  Skipping {collection.id}: no items")
426        return None
427
428    # Determine hemisphere if not provided
429    if hemisphere is None:
430        # Try from collection name first
431        hemisphere = determine_hemisphere_from_name(collection.id)
432
433        # Fall back to geometry-based detection
434        if hemisphere is None:
435            hemisphere = determine_hemisphere_from_geometry(collection_items)
436            if verbose and hemisphere:
437                print(f"  Detected hemisphere from geometry: {hemisphere}")
438
439    # Get provider from config if not provided
440    if provider is None:
441        provider = config.data.get('provider')
442
443    if verbose:
444        if hemisphere:
445            print(f"  Hemisphere: {hemisphere}")
446        else:
447            print(f"  WARNING: Could not determine hemisphere for {collection.id}")
448        if provider:
449            print(f"  Provider: {provider}")
450        else:
451            print(f"  WARNING: No provider specified for {collection.id}")
452
453    # Ensure output directory exists
454    output_dir.mkdir(parents=True, exist_ok=True)
455
456    # Export to parquet
457    parquet_file = output_dir / f"{collection.id}.parquet"
458
459    if verbose:
460        print(f"  Exporting collection: {collection.id} ({len(collection_items)} items)")
461
462    # Compute morton polygon from all collection items
463    mpolygon = compute_mpolygon_from_items(collection_items)
464
465    # Build collections metadata - single collection in this case
466    collection_dict = collection.to_dict()
467
468    # Add OPR extension to collection
469    collection_exts = collection_dict.setdefault('stac_extensions', [])
470    if OPR_EXT not in collection_exts:
471        collection_exts.append(OPR_EXT)
472
473    # Add OPR metadata to collection
474    props = collection_dict.setdefault('properties', {})
475    props['opr:mpolygon'] = mpolygon
476    if hemisphere:
477        props['opr:hemisphere'] = hemisphere
478    if provider:
479        props['opr:provider'] = provider
480
481    # Clean collection links - remove item links with None hrefs
482    if 'links' in collection_dict:
483        collection_dict['links'] = [
484            link for link in collection_dict['links']
485            if not (link.get('rel') == 'item' and link.get('href') is None)
486        ]
487    # Clean items and add xopr metadata before export
488    clean_items = []
489    for item in collection_items:
490        item_dict = item.to_dict()
491
492        # Add OPR metadata to each item
493        if 'properties' not in item_dict:
494            item_dict['properties'] = {}
495        if hemisphere:
496            item_dict['properties']['opr:hemisphere'] = hemisphere
497        if provider:
498            item_dict['properties']['opr:provider'] = provider
499
500        # Clean links with None hrefs
501        if 'links' in item_dict:
502            item_dict['links'] = [
503                link for link in item_dict['links']
504                if link.get('href') is not None
505            ]
506        clean_items.append(item_dict)
507
508    # Convert items to Arrow format
509    record_batch_reader = stac_geoparquet.arrow.parse_stac_items_to_arrow(clean_items)
510
511    # When all items have empty links arrays, PyArrow infers the column type
512    # as list<null> which downstream tools (e.g. stac-wasm) cannot
513    # deserialize. Cast to the correct STAC link struct type.
514    table = record_batch_reader.read_all()
515
516    # Enforce int type for opr:frequency/opr:bandwidth per OPR extension schema.
517    # Inferred from item values — fail loudly if any item slipped through as float.
518    for field_name in ('opr:frequency', 'opr:bandwidth'):
519        idx = table.schema.get_field_index(field_name)
520        if idx != -1 and not pa.types.is_integer(table.schema.field(idx).type):
521            raise ValueError(
522                f"{field_name} must be integer per OPR extension schema, got "
523                f"{table.schema.field(idx).type}"
524            )
525
526    links_field = table.schema.field('links')
527    if pa.types.is_null(links_field.type.value_type):
528        links_type = pa.list_(pa.struct([
529            ('href', pa.string()),
530            ('rel', pa.string()),
531            ('type', pa.string()),
532        ]))
533        links_idx = table.schema.get_field_index('links')
534        table = table.set_column(
535            links_idx,
536            pa.field('links', links_type),
537            table.column('links').cast(links_type),
538        )
539
540    # Write to Parquet with collection metadata
541    # Note: Using collection_metadata for compatibility with stac-geoparquet 0.7.0
542    # In newer versions (>0.8), this should be 'collections' parameter
543    stac_geoparquet.arrow.to_parquet(
544        table=table,
545        output_path=parquet_file,
546        collection_metadata=collection_dict,  # Single collection metadata (cleaned)
547        schema_version="1.1.0",  # Use latest schema version
548        compression="snappy",  # Use snappy compression for better performance
549        write_statistics=True  # Write column statistics for query optimization
550    )
551
552    if verbose:
553        size_kb = parquet_file.stat().st_size / 1024
554        print(f"  ✅ {collection.id}.parquet saved ({size_kb:.1f} KB)")
555
556    return parquet_file

 93def extract_item_metadata(
 94    mat_file_path: Union[str, Path] = None,
 95    dataset=None,
 96    conf: Optional[DictConfig] = None
 97) -> Dict[str, Any]:
 98    """
 99    Extract metadata from MAT/HDF5 file with optional configuration.
100
101    Parameters
102    ----------
103    mat_file_path : Union[str, Path], optional
104        Path or URL to MAT/HDF5 file
105    dataset : xarray.Dataset, optional
106        Pre-loaded dataset
107    conf : DictConfig, optional
108        Configuration for geometry simplification
109
110    Returns
111    -------
112    Dict[str, Any]
113        Extracted metadata including geometry, bbox, date, etc.
114    """
115    # Validate input
116    if (mat_file_path is None) == (dataset is None):
117        raise ValueError("Exactly one of mat_file_path or dataset must be provided")
118
119    should_close_dataset = False
120
121    if mat_file_path is not None:
122        if isinstance(mat_file_path, str):
123            file_path = Path(mat_file_path)
124        else:
125            file_path = mat_file_path
126
127        # Check existence for local files
128        if not str(mat_file_path).startswith(('http://', 'https://')):
129            if not file_path.exists():
130                raise FileNotFoundError(f"MAT file not found: {file_path}")
131
132        opr = OPRConnection(cache_dir=None)
133        ds = opr.load_frame_url(str(mat_file_path))
134        should_close_dataset = True
135    else:
136        ds = dataset
137
138    with warnings.catch_warnings():
139        warnings.simplefilter("ignore", UserWarning)
140        date = pd.to_datetime(ds['slow_time'].mean().values).to_pydatetime()
141
142    # Create geometry
143    geom_series = gpd.GeoSeries(map(Point, zip(ds['Longitude'].values, ds['Latitude'].values)))
144    raw_line = LineString(geom_series.tolist())
145
146    # Apply simplification based on config
147    line = raw_line
148    if conf and conf.get('geometry', {}).get('simplify', True):
149        tolerance = conf.geometry.get('tolerance', 100.0)
150        line = simplify_geometry_polar_projection(line, simplify_tolerance=tolerance)
151
152    bounds = shapely.bounds(line)
153    boundingbox = box(bounds[0], bounds[1], bounds[2], bounds[3])
154
155    # Extract radar parameters with config fallback support
156    low_freq = None
157    high_freq = None
158    used_config_fallback = False
159
160    # Check if config provides radar frequency values
161    config_has_radar = (
162        conf is not None
163        and conf.get('radar', {}).get('f0') is not None
164        and conf.get('radar', {}).get('f1') is not None
165    )
166    config_override = conf is not None and conf.get('radar', {}).get('override', False)
167
168    if config_override and config_has_radar:
169        # Config override takes precedence
170        low_freq = float(conf.radar.f0)
171        high_freq = float(conf.radar.f1)
172    else:
173        # Try to extract from data first
174        try:
175            stable_wfs = extract_stable_wfs_params(find_radar_wfs_params(ds))
176            if 'f0' in stable_wfs and 'f1' in stable_wfs:
177                low_freq_array = stable_wfs['f0']
178                high_freq_array = stable_wfs['f1']
179
180                unique_low_freq = np.unique(low_freq_array)
181                if len(unique_low_freq) != 1:
182                    raise ValueError(f"Multiple low frequency values found: {unique_low_freq}")
183                low_freq = float(unique_low_freq[0])
184
185                unique_high_freq = np.unique(high_freq_array)
186                if len(unique_high_freq) != 1:
187                    raise ValueError(f"Multiple high frequency values found: {unique_high_freq}")
188                high_freq = float(unique_high_freq[0])
189        except KeyError:
190            pass  # Will try config fallback
191
192        # Config fallback if extraction failed
193        if (low_freq is None or high_freq is None) and config_has_radar:
194            low_freq = float(conf.radar.f0)
195            high_freq = float(conf.radar.f1)
196            used_config_fallback = True
197
198    # Error if neither source has values
199    if low_freq is None or high_freq is None:
200        raise ValueError(
201            "Radar frequency parameters (f0, f1) not found in data file "
202            "and not provided in config. Add radar.f0 and radar.f1 to your config."
203        )
204
205    # Log warning if using fallback (only when verbose)
206    if used_config_fallback and conf and conf.get('logging', {}).get('verbose', False):
207        logging.warning(f"Using config fallback for radar frequencies: f0={low_freq}, f1={high_freq}")
208
209    bandwidth = float(np.abs(high_freq - low_freq))
210    center_freq = float((low_freq + high_freq) / 2)
211
212    # Extract science metadata with config fallback support
213    doi = None
214    cite = None
215    used_sci_fallback = False
216
217    # Check if config provides sci metadata values
218    config_has_sci = conf is not None and conf.get('sci') is not None
219    sci_override = config_has_sci and conf.get('sci', {}).get('override', False)
220
221    if sci_override and config_has_sci:
222        # Config override takes precedence
223        doi = conf.sci.get('doi')
224        cite = conf.sci.get('citation')
225    else:
226        # Try to extract from data first
227        doi = ds.attrs.get('doi', None)
228        cite = ds.attrs.get('funder_text', None)
229
230        # Config fallback if extraction returned None
231        if config_has_sci:
232            if doi is None and conf.sci.get('doi') is not None:
233                doi = conf.sci.doi
234                used_sci_fallback = True
235            if cite is None and conf.sci.get('citation') is not None:
236                cite = conf.sci.citation
237                used_sci_fallback = True
238
239    # Log warning if using fallback (only when verbose)
240    if used_sci_fallback and conf and conf.get('logging', {}).get('verbose', False):
241        logging.warning(f"Using config fallback for sci metadata: doi={doi}, citation={cite}")
242
243    mime = ds.attrs['mimetype']
244
245    if should_close_dataset:
246        ds.close()
247
248    return {
249        'geom': line,
250        'raw_geom': raw_line,
251        'bbox': boundingbox,
252        'date': date,
253        'frequency': center_freq,
254        'bandwidth': bandwidth,
255        'doi': doi,
256        'citation': cite,
257        'mimetype': mime
258    }

261def discover_campaigns(data_root: Union[str, Path], conf: Optional[DictConfig] = None) -> List[Dict[str, str]]:
262    """
263    Discover all campaigns in the data directory.
264
265    Parameters
266    ----------
267    data_root : Union[str, Path]
268        Root directory containing campaign subdirectories
269    conf : DictConfig, optional
270        Configuration with optional filters
271
272    Returns
273    -------
274    List[Dict[str, str]]
275        List of campaign metadata dictionaries
276    """
277    campaign_pattern = re.compile(r'^(\d{4})_([^_]+)_(.+)$')
278    campaigns = []
279
280    data_root = Path(data_root)
281
282    if not data_root.exists():
283        raise FileNotFoundError(f"Data root directory not found: {data_root}")
284
285    for item in data_root.iterdir():
286        if item.is_dir():
287            match = campaign_pattern.match(item.name)
288            if match:
289                year, location, aircraft = match.groups()
290
291                # Apply filters if config provided
292                if conf and 'campaigns' in conf.data:
293                    include = conf.data.campaigns.get('include', [])
294                    exclude = conf.data.campaigns.get('exclude', [])
295
296                    if include and item.name not in include:
297                        continue
298                    if exclude and item.name in exclude:
299                        continue
300
301                campaigns.append({
302                    'name': item.name,
303                    'year': year,
304                    'location': location,
305                    'aircraft': aircraft,
306                    'path': str(item)
307                })
308
309    return sorted(campaigns, key=lambda x: (x['year'], x['name']))

27def discover_flight_lines(campaign_path: Union[str, Path], conf: DictConfig) -> List[Dict[str, Any]]:
28    """
29    Discover flight lines for a campaign using configuration.
30
31    Parameters
32    ----------
33    campaign_path : Union[str, Path]
34        Path to campaign directory
35    conf : DictConfig
36        Configuration object with data.primary_product and data.extra_products
37
38    Returns
39    -------
40    List[Dict[str, Any]]
41        List of flight line metadata dictionaries
42    """
43    campaign_path = Path(campaign_path)
44
45    # Get products from config
46    primary_product = conf.data.primary_product
47    extra_products = conf.data.get('extra_products', []) or []
48
49    product_path = campaign_path / primary_product
50
51    if not product_path.exists():
52        raise FileNotFoundError(f"Data product directory not found: {product_path}")
53
54    flight_pattern = re.compile(r'^(\d{8}_\d+)$')
55    flights = []
56
57    for flight_dir in product_path.iterdir():
58        if flight_dir.is_dir():
59            match = flight_pattern.match(flight_dir.name)
60            if match:
61                flight_id = match.group(1)
62                parts = flight_id.split('_')
63                date_part = parts[0]
64                flight_num = parts[1]
65
66                # Collect data files for primary product
67                data_files = {
68                    primary_product: {
69                        f.name: str(f) for f in flight_dir.glob("*.mat")
70                        if "_img" not in f.name
71                    }
72                }
73
74                # Include extra data products if they exist
75                for extra_product in extra_products:
76                    extra_product_path = campaign_path / extra_product / flight_dir.name
77                    if extra_product_path.exists():
78                        data_files[extra_product] = {
79                            f.name: str(f) for f in extra_product_path.glob("*.mat")
80                        }
81
82                if data_files:
83                    flights.append({
84                        'flight_id': flight_id,
85                        'date': date_part,
86                        'flight_num': flight_num,
87                        'data_files': data_files
88                    })
89
90    return sorted(flights, key=lambda x: x['flight_id'])

355def collect_uniform_metadata(items: List, property_keys: List[str]) -> tuple[List[str], dict]:
356    """
357    Collect metadata properties that have uniform values across items.
358
359    Parameters
360    ----------
361    items : List[pystac.Item]
362        List of STAC items to extract metadata from
363    property_keys : List[str]
364        List of property keys to check
365
366    Returns
367    -------
368    tuple
369        (extensions_needed, extra_fields_dict)
370    """
371    SCI_EXT = 'https://stac-extensions.github.io/scientific/v1.0.0/schema.json'
372    SAR_EXT = 'https://stac-extensions.github.io/sar/v1.3.0/schema.json'
373
374    extensions = []
375    extra_fields = {}
376
377    property_mappings = {
378        'sci:doi': SCI_EXT,
379        'sci:citation': SCI_EXT,
380        'sar:center_frequency': SAR_EXT,
381        'sar:bandwidth': SAR_EXT
382    }
383
384    for key in property_keys:
385        values = [
386            item.properties.get(key)
387            for item in items
388            if item.properties.get(key) is not None
389        ]
390
391        if values and len(np.unique(values)) == 1:
392            ext = property_mappings.get(key)
393            if ext and ext not in extensions:
394                extensions.append(ext)
395            extra_fields[key] = values[0]
396
397    return extensions, extra_fields

 68def build_collection_extent_and_geometry(
 69    items: List[pystac.Item]
 70) -> pystac.Extent:
 71    """
 72    Calculate spatial and temporal extent from a list of items.
 73
 74    Parameters
 75    ----------
 76    items : list of pystac.Item
 77        List of STAC items to compute extent from.
 78
 79    Returns
 80    -------
 81    pystac.Extent
 82        Combined spatial and temporal extent covering all input items.
 83
 84    Raises
 85    ------
 86    ValueError
 87        If items list is empty.
 88    """
 89    if not items:
 90        raise ValueError("Cannot build extent from empty item list")
 91
 92    # Build extent using bboxes
 93    bboxes = []
 94    datetimes = []
 95
 96    for item in items:
 97        if item.bbox:
 98            bbox_geom = shapely.geometry.box(*item.bbox)
 99            bboxes.append(bbox_geom)
100
101        if item.datetime:
102            datetimes.append(item.datetime)
103
104    if bboxes:
105        union_bbox = bboxes[0]
106        for bbox in bboxes[1:]:
107            union_bbox = union_bbox.union(bbox)
108
109        collection_bbox = list(union_bbox.bounds)
110        spatial_extent = pystac.SpatialExtent(bboxes=[collection_bbox])
111    else:
112        spatial_extent = pystac.SpatialExtent(bboxes=[[-180, -90, 180, 90]])
113
114    if datetimes:
115        sorted_times = sorted(datetimes)
116        temporal_extent = pystac.TemporalExtent(
117            intervals=[[sorted_times[0], sorted_times[-1]]]
118        )
119    else:
120        temporal_extent = pystac.TemporalExtent(intervals=[[None, None]])
121
122    extent = pystac.Extent(spatial=spatial_extent, temporal=temporal_extent)
123    return extent