COPC Operations

class CopcReader:
    def __init__(self, stream, close_fd=True, http_num_threads=8, decompression_selection=None):
        """
        Initialize COPC reader.
        
        Parameters:
        - stream: BinaryIO - COPC file stream
        - close_fd: bool - Whether to close file descriptor (default: True)
        - http_num_threads: int - Number of HTTP worker threads (default: 8)
        - decompression_selection: DecompressionSelection - Fields to decompress
        """
    
    @classmethod
    def open(cls, source, http_num_threads=8, decompression_selection=None) -> CopcReader:
        """
        Open COPC file from local path or HTTP URL.
        
        Parameters:
        - source: str, Path, or URL - COPC file location
        - http_num_threads: int - Number of HTTP worker threads for remote files
        - decompression_selection: DecompressionSelection - Fields to decompress
        
        Returns:
        CopcReader: COPC reader instance
        """
    
    @property
    def header(self) -> LasHeader:
        """Get LAS header for the COPC file."""
    
    def query(self, bounds=None, resolution=None, level=None) -> ScaleAwarePointRecord:
        """
        General query method supporting spatial bounds, resolution, and level constraints.
        
        Parameters:
        - bounds: Bounds - Spatial bounds for query (optional)
        - resolution: float - Target point resolution/density (optional)
        - level: int or range - Octree level(s) to query (optional)
        
        Returns:
        ScaleAwarePointRecord: Points matching query criteria
        """
    
    def spatial_query(self, bounds: Bounds) -> ScaleAwarePointRecord:
        """
        Query points within specified spatial bounds.
        
        Parameters:
        - bounds: Bounds - Spatial query bounds
        
        Returns:
        ScaleAwarePointRecord: Points within bounds
        """
    
    def level_query(self, level) -> ScaleAwarePointRecord:
        """
        Query points at specific octree level(s).
        
        Parameters:
        - level: int or range - Octree level or level range
        
        Returns:
        ScaleAwarePointRecord: Points at specified level(s)
        """
    
    def close(self):
        """Close COPC reader and free resources."""
    
    def __enter__(self) -> CopcReader: ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

import laspy
from laspy import CopcReader, Bounds

# Open local COPC file
with CopcReader.open('data.copc.laz') as reader:
    print(f"COPC file has {reader.header.point_count} points")
    
    # Query all points
    all_points = reader.query()
    print(f"Retrieved {len(all_points)} points")

# Open remote COPC file via HTTP
url = "https://example.com/data.copc.laz"
with CopcReader.open(url, http_num_threads=4) as reader:
    # Spatial query
    bounds = Bounds(
        mins=[100, 200, 0], 
        maxs=[200, 300, 50]
    )
    spatial_points = reader.spatial_query(bounds)
    print(f"Spatial query returned {len(spatial_points)} points")

# Level-based query for different detail levels
with CopcReader.open('data.copc.laz') as reader:
    # Get coarse overview (low levels)
    overview = reader.level_query(range(0, 3))
    
    # Get detailed data (high levels) 
    detailed = reader.level_query(range(8, 12))
    
    print(f"Overview: {len(overview)} points")
    print(f"Detailed: {len(detailed)} points")

class Bounds:
    def __init__(self, mins, maxs):
        """
        Create spatial bounds.
        
        Parameters:
        - mins: array-like - Minimum coordinates [x, y] or [x, y, z]
        - maxs: array-like - Maximum coordinates [x, y] or [x, y, z]
        """
    
    @property
    def mins(self) -> np.ndarray:
        """Minimum coordinate values."""
    
    @property
    def maxs(self) -> np.ndarray:
        """Maximum coordinate values."""
    
    def overlaps(self, other: Bounds) -> bool:
        """
        Check if bounds overlap with another bounds.
        
        Parameters:
        - other: Bounds - Other bounds to check against
        
        Returns:
        bool: True if bounds overlap
        """
    
    @staticmethod
    def ensure_3d(mins, maxs) -> Bounds:
        """
        Ensure bounds are 3D by extending 2D bounds.
        
        Parameters:
        - mins: array-like - Minimum coordinates
        - maxs: array-like - Maximum coordinates  
        
        Returns:
        Bounds: 3D bounds object
        """

import numpy as np
from laspy import Bounds

# Create 2D bounds (Z dimension will be unlimited)
bounds_2d = Bounds(
    mins=[1000, 2000],
    maxs=[1500, 2500]
)

# Create 3D bounds
bounds_3d = Bounds(
    mins=[1000, 2000, 100],
    maxs=[1500, 2500, 200]
)

# Check bounds properties
print(f"2D bounds mins: {bounds_2d.mins}")
print(f"2D bounds maxs: {bounds_2d.maxs}")

# Check overlap
other_bounds = Bounds([1400, 2400], [1600, 2600])
overlaps = bounds_2d.overlaps(other_bounds)
print(f"Bounds overlap: {overlaps}")

# Convert 2D to 3D bounds
bounds_3d_converted = Bounds.ensure_3d([1000, 2000], [1500, 2500])
print(f"3D converted: {bounds_3d_converted.mins} to {bounds_3d_converted.maxs}")

import laspy
from laspy import CopcReader, Bounds

def query_multiple_regions(copc_file, regions):
    """Query multiple spatial regions efficiently."""
    with CopcReader.open(copc_file) as reader:
        all_points = []
        
        for region_name, (mins, maxs) in regions.items():
            bounds = Bounds(mins, maxs)
            points = reader.spatial_query(bounds)
            print(f"Region '{region_name}': {len(points)} points")
            all_points.append(points)
        
        return all_points

# Define regions of interest
regions = {
    "downtown": ([1000, 2000, 0], [1200, 2200, 100]),
    "park": ([1500, 2500, 0], [1700, 2700, 50]),
    "industrial": ([2000, 3000, 0], [2300, 3300, 80])
}

results = query_multiple_regions('city.copc.laz', regions)

import laspy
from laspy import CopcReader

def progressive_query(copc_file, max_points=1000000):
    """Query with progressive level-of-detail until point limit reached."""
    with CopcReader.open(copc_file) as reader:
        level = 0
        total_points = 0
        
        while total_points < max_points:
            level_points = reader.level_query(level)
            if len(level_points) == 0:
                break
                
            total_points += len(level_points)
            print(f"Level {level}: {len(level_points)} points (total: {total_points})")
            
            # Process level points
            yield level, level_points
            level += 1

# Use progressive querying  
for level, points in progressive_query('large.copc.laz'):
    # Process points at this level of detail
    density = len(points) / (points.x.max() - points.x.min()) / (points.y.max() - points.y.min())
    print(f"Level {level} density: {density:.2f} points/unit²")

import laspy
from laspy import CopcReader, DecompressionSelection

# Optimize for network access
def stream_copc_efficiently(url, regions, num_threads=16):
    """Stream COPC data with optimized HTTP settings."""
    
    # Use selective decompression to reduce bandwidth
    selection = DecompressionSelection.base()  # Only essential fields
    
    with CopcReader.open(url, 
                        http_num_threads=num_threads,
                        decompression_selection=selection) as reader:
        
        print(f"Streaming from: {url}")
        print(f"File bounds: {reader.header.mins} to {reader.header.maxs}")
        
        for region_name, bounds in regions.items():
            print(f"Fetching region: {region_name}")
            points = reader.spatial_query(bounds)
            
            # Process immediately to minimize memory usage
            ground_points = points[points.classification == 2]
            yield region_name, ground_points

# Stream from cloud storage
url = "https://cloud-storage.example.com/lidar/city.copc.laz"
regions = {
    "area1": Bounds([1000, 2000], [1100, 2100]),
    "area2": Bounds([1200, 2200], [1300, 2300])
}

for region_name, ground_points in stream_copc_efficiently(url, regions):
    print(f"Processing {len(ground_points)} ground points from {region_name}")

import laspy
from laspy import CopcReader, Bounds

# Query COPC data and write to standard LAS
with CopcReader.open('input.copc.laz') as reader:
    # Query region of interest
    bounds = Bounds([1000, 2000, 0], [1500, 2500, 100])
    points = reader.spatial_query(bounds)
    
    # Create LAS data container
    las = laspy.LasData(reader.header, points)
    
    # Apply filters
    las.points = las.points[las.classification.isin([2, 3, 4])]  # Ground, low/med vegetation
    
    # Write filtered data
    las.write('filtered_region.las')

# Convert COPC query results to different point format
with CopcReader.open('input.copc.laz') as reader:
    points = reader.query()
    las = laspy.LasData(reader.header, points)
    
    # Convert to point format with RGB
    converted = laspy.convert(las, point_format_id=7)
    converted.write('converted.las')

import laspy
from laspy import CopcReader, Bounds
from laspy.errors import LaspyException

try:
    with CopcReader.open('https://example.com/missing.copc.laz') as reader:
        points = reader.spatial_query(Bounds([0, 0], [100, 100]))
except LaspyException as e:
    print(f"COPC error: {e}")
except Exception as e:
    print(f"Network or other error: {e}")