quantum/config/hdf5parser.py

import h5py
import numpy as np
from pathlib import Path

def load_map_from_hdf5(h5_source):
    """
    Load map data from HDF5 file. Accepts:
        - string path (e.g., "map.h5")
        - file-like object (e.g., SpooledTemporaryFile)
    
    Returns structured dict.
    """
    with h5py.File(h5_source, 'r') as f:
        # Load required data
        occupancy = f['map_structure'][:]
        M, N = occupancy.shape
        
        # Extract obstacles
        obstacles = []
        obs_positions = np.where(occupancy == 1)
        for i, j in zip(obs_positions[0], obs_positions[1]):
            obstacles.append((int(i), int(j)))

        # Load metadata
        map_name = f.attrs.get('map_name', None)
        
        # Only use filename stem if map_name not set AND source is a path
        # Note that path stem was giving problem with fastApi
        if map_name is None:
            # Try to get filename only if h5_source is a string or has name
            try:
                if isinstance(h5_source, (str, Path)):
                    map_name = Path(h5_source).stem
                elif hasattr(h5_source, 'name') and isinstance(h5_source.name, str):
                    map_name = Path(h5_source.name).stem
                else:
                    map_name = "unknown_map"
            except Exception:
                map_name = "unknown_map"

        resolution = f.attrs.get('resolution', 1.0)

        # Load materials
        materials = []
        if 'materials' in f:
            materials = [mat.decode('utf-8') for mat in f['materials'][:]]

        result = {
            'name': map_name,
            'grid': {
                'M': M,
                'N': N,
                'obstacles': obstacles
            },
            'resolution': resolution,
            'materials': materials
        }

        if 'terrain' in f:
            result['terrain_grid'] = f['terrain'][:]
        if 'elevation' in f:
            result['elevation_grid'] = f['elevation'][:]

        return result


def load_graph_from_hdf5(h5_source):
    with h5py.File(h5_source, "r") as f:
        nodes = f["graph/nodes"][:]   # shape (N, 2)
        edges = f["graph/edges"][:]   # shape (E, 3)

        # Load metadata
        map_name = f.attrs.get('map_name', None)
        
        # Only use filename stem if map_name not set AND source is a path
        # Note that path stem was giving problem with fastApi
        if map_name is None:
            # Try to get filename only if h5_source is a string or has name
            try:
                if isinstance(h5_source, (str, Path)):
                    map_name = Path(h5_source).stem
                elif hasattr(h5_source, 'name') and isinstance(h5_source.name, str):
                    map_name = Path(h5_source.name).stem
                else:
                    map_name = "unknown_map"
            except Exception:
                map_name = "unknown_map"

        resolution = f.attrs.get('resolution', 1.0)

        result = {
            'name': map_name,
            'edges': edges,
            'nodes': nodes,
            'resolution': resolution,
        }

        return result

def load_both_from_hdf5(h5_source):
    """
    Load both map (grid) and graph data from HDF5 file. 
    This is useful for synthetic maps that support both approaches.
    
    Args:
        h5_source: HDF5 file path or file-like object
        
    Returns:
        dict: Contains both 'map_data' and 'graph_data' keys, plus metadata
    """
    with h5py.File(h5_source, 'r') as f:
        # Load common metadata
        map_name = f.attrs.get('map_name', None)
        
        # Only use filename stem if map_name not set AND source is a path
        if map_name is None:
            try:
                if isinstance(h5_source, (str, Path)):
                    map_name = Path(h5_source).stem
                elif hasattr(h5_source, 'name') and isinstance(h5_source.name, str):
                    map_name = Path(h5_source.name).stem
                else:
                    map_name = "unknown_map"
            except Exception:
                map_name = "unknown_map"

        resolution = f.attrs.get('resolution', 1.0)
        
        # Check what data is available
        has_map = 'map_structure' in f
        has_graph = 'graph' in f
        
        result = {
            'name': map_name,
            'resolution': resolution,
            'has_map': has_map,
            'has_graph': has_graph,
            'map_data': None,
            'graph_data': None
        }
        
        # Load map data if available
        if has_map:
            occupancy = f['map_structure'][:]
            M, N = occupancy.shape
            
            # Extract obstacles
            obstacles = []
            obs_positions = np.where(occupancy == 1)
            for i, j in zip(obs_positions[0], obs_positions[1]):
                obstacles.append((int(i), int(j)))

            # Load materials
            materials = []
            if 'materials' in f:
                materials = [mat.decode('utf-8') for mat in f['materials'][:]]

            map_data = {
                'name': map_name,
                'grid': {
                    'M': M,
                    'N': N,
                    'obstacles': obstacles
                },
                'resolution': resolution,
                'materials': materials
            }

            if 'terrain' in f:
                map_data['terrain_grid'] = f['terrain'][:]
            if 'elevation' in f:
                map_data['elevation_grid'] = f['elevation'][:]
                
            result['map_data'] = map_data
        
        # Load graph data if available
        if has_graph:
            nodes = f["graph/nodes"][:]   # shape (N, 2)
            edges = f["graph/edges"][:]   # shape (E, 3)

            graph_data = {
                'name': map_name,
                'edges': edges,
                'nodes': nodes,
                'resolution': resolution,
            }
            
            result['graph_data'] = graph_data

        return result

    
# Example usage
if __name__ == "__main__":
    import sys
    h5_path = sys.argv[1] if len(sys.argv) > 1 else "quantum/maps/synthetic/3x3/no_obs3x3.h5"
    
    # Example 1: Load both map and graph data
    print("=== Loading both map and graph data ===")
    both_data = load_both_from_hdf5(h5_path)
    print(f"Name: {both_data['name']}")
    print(f"Has map data: {both_data['has_map']}")
    print(f"Has graph data: {both_data['has_graph']}")
    print(f"Resolution: {both_data['resolution']}")
    
    if both_data['map_data']:
        print("\nMap data:")
        print(f"Grid: {both_data['map_data']['grid']['M']}x{both_data['map_data']['grid']['N']}")
        print(f"Obstacles: {both_data['map_data']['grid']['obstacles']}")
        if 'terrain_grid' in both_data['map_data']:
            print("Terrain grid loaded (shape):", both_data['map_data']['terrain_grid'].shape)
        if 'elevation_grid' in both_data['map_data']:
            print("Elevation grid loaded (shape):", both_data['map_data']['elevation_grid'].shape)
    
    if both_data['graph_data']:
        print("\nGraph data:")
        print(f"Nodes shape: {both_data['graph_data']['nodes'].shape}")
        print(f"Edges shape: {both_data['graph_data']['edges'].shape}")
    
    # Example 2: Load only map data (backward compatibility)
    print("\n=== Loading only map data (backward compatibility) ===")
    map_data = load_map_from_hdf5(h5_path)
    print(f"Name: {map_data['name']}")
    print(f"Grid: {map_data['grid']['M']}x{map_data['grid']['N']}")
    print(f"Obstacles: {map_data['grid']['obstacles']}")
    print(f"Resolution: {map_data['resolution']}")
    
    # Example 3: Load only graph data (backward compatibility)
    print("\n=== Loading only graph data (backward compatibility) ===")
    graph_data = load_graph_from_hdf5(h5_path)
    print(f"Name: {graph_data['name']}")
    print(f"Nodes shape: {graph_data['nodes'].shape}")
    print(f"Edges shape: {graph_data['edges'].shape}")
    print(f"Resolution: {graph_data['resolution']}")