StoutLoader

StoutLoader

StoutLoader()

Data loader for STOUT campaign management system.

Provides methods to load flight data paths and associated metadata from the STOUT database and file system.

Attributes:

Name	Type	Description
campaign_service	Optional[CampaignService]	Service for accessing campaign and flight data
base_data_path	Optional[Path]	Base path where all campaign data is stored

Initialize the StoutDataLoader.

Initializes the loader and attempts to connect to stout campaign service. Falls back to filesystem queries if stout import fails.

Source code in pils/loader/stout.py

def __init__(self):
    """
    Initialize the StoutDataLoader.

    Initializes the loader and attempts to connect to stout campaign service.
    Falls back to filesystem queries if stout import fails.
    """

    self.campaign_service = None

    try:
        from stout.config import Config  # type: ignore
        from stout.services.campaigns import CampaignService  # type: ignore

        self.campaign_service = CampaignService()
        self.base_data_path = Config.MAIN_DATA_PATH
        logger.info(
            f"Initialized with stout database, base path: {self.base_data_path}"
        )
    except ImportError as e:
        logger.warning(
            f"Could not import stout: {e}. Falling back to filesystem queries."
        )
        self.use_stout = False

load_all_flights

load_all_flights() -> list[dict[str, Any]]

Load all flights from all campaigns.

Returns:

Type	Description
list[dict[str, Any]]	list of flight dictionaries containing flight metadata and paths. Each flight dict includes: flight_id, flight_name, campaign_id, takeoff_datetime, landing_datetime, and folder paths.

Source code in pils/loader/stout.py

def load_all_flights(self) -> list[dict[str, Any]]:
    """
    Load all flights from all campaigns.

    Returns
    -------
    list[dict[str, Any]]
        list of flight dictionaries containing flight metadata and paths.
        Each flight dict includes: flight_id, flight_name, campaign_id,
        takeoff_datetime, landing_datetime, and folder paths.
    """
    logger.info("Loading all flights from all campaigns...")

    if self.campaign_service is None:
        raise RuntimeError("Campaign service not initialized")
    try:
        flights = self.campaign_service.get_all_flights()
        logger.info(f"Loaded {len(flights)} flights from database")
        return flights
    except Exception as e:
        logger.error(f"Error loading flights from database: {e}")
        raise

load_all_campaign_flights

load_all_campaign_flights(campaign_id: str | None = None, campaign_name: str | None = None) -> list[dict[str, Any]] | None

Load all flights from a specific campaign.

Parameters:

Name	Type	Description	Default
campaign_id	Optional[str]	Campaign ID to load	None
campaign_name	Optional[str]	Campaign name to load (alternative to campaign_id)	None

Returns:

Type	Description
Optional[dict[str, Any]]	Campaign dictionary with metadata and paths, or None if not found.

Source code in pils/loader/stout.py

def load_all_campaign_flights(
    self, campaign_id: str | None = None, campaign_name: str | None = None
) -> list[dict[str, Any]] | None:
    """
    Load all flights from a specific campaign.

    Parameters
    ----------
    campaign_id : Optional[str]
        Campaign ID to load
    campaign_name : Optional[str]
        Campaign name to load (alternative to campaign_id)

    Returns
    -------
    Optional[dict[str, Any]]
        Campaign dictionary with metadata and paths, or None if not found.
    """
    if not campaign_id and not campaign_name:
        raise ValueError("Either flight_id or flight_name must be provided")

    logger.info(
        f"Loading single flight: flight_id={campaign_id}, flight_name={campaign_name}"
    )

    if self.campaign_service is None:
        raise RuntimeError("Campaign service not initialized")
    try:
        # Get the campaign_id first if only campaign_name provided
        cid = campaign_id or campaign_name

        if not cid:
            raise ValueError("Either campaign_id or campaign_name must be provided")

        flights = self.campaign_service.get_flights_by_campaign(campaign_id=cid)

        for flight in flights:
            if flight:
                logger.info(f"Loaded flight: {flight.get('flight_name')}")
        return flights
    except Exception as e:
        logger.error(f"Error loading flight from database: {e}")
        raise

load_single_flight

load_single_flight(flight_id: str | None = None, flight_name: str | None = None) -> dict[str, Any] | None

Load data for a single flight.

Parameters:

Name	Type	Description	Default
flight_id	Optional[str]	Flight ID to load	None
flight_name	Optional[str]	Flight name to load (alternative to flight_id)	None

Returns:

Type	Description
Optional[dict[str, Any]]	Flight dictionary with metadata and paths, or None if not found.

Source code in pils/loader/stout.py

def load_single_flight(
    self, flight_id: str | None = None, flight_name: str | None = None
) -> dict[str, Any] | None:
    """
    Load data for a single flight.

    Parameters
    ----------
    flight_id : Optional[str]
        Flight ID to load
    flight_name : Optional[str]
        Flight name to load (alternative to flight_id)

    Returns
    -------
    Optional[dict[str, Any]]
        Flight dictionary with metadata and paths, or None if not found.
    """
    if not flight_id and not flight_name:
        raise ValueError("Either flight_id or flight_name must be provided")

    logger.info(
        f"Loading single flight: flight_id={flight_id}, flight_name={flight_name}"
    )

    if self.campaign_service is None:
        raise RuntimeError("Campaign service not initialized")
    try:
        flight = self.campaign_service.get_flight(
            flight_name=flight_name, flight_id=flight_id
        )
        if flight:
            logger.info(f"Loaded flight: {flight.get('flight_name')}")
        return flight
    except Exception as e:
        logger.error(f"Error loading flight from database: {e}")
        raise

load_flights_by_date

load_flights_by_date(start_date: str, end_date: str, campaign_id: str | None = None) -> list[dict[str, Any]]

Load flights within a date range.

Parameters:

Name	Type	Description	Default
start_date	str	Start date in format 'YYYY-MM-DD'	required
end_date	str	End date in format 'YYYY-MM-DD'	required
campaign_id	Optional[str]	Filter by campaign ID (optional)	None

Returns:

Type	Description
list[dict[str, Any]]	list of flight dictionaries matching the date range.

Source code in pils/loader/stout.py

def load_flights_by_date(
    self, start_date: str, end_date: str, campaign_id: str | None = None
) -> list[dict[str, Any]]:
    """
    Load flights within a date range.

    Parameters
    ----------
    start_date : str
        Start date in format 'YYYY-MM-DD'
    end_date : str
        End date in format 'YYYY-MM-DD'
    campaign_id : Optional[str]
        Filter by campaign ID (optional)

    Returns
    -------
    list[dict[str, Any]]
        list of flight dictionaries matching the date range.
    """
    try:
        start_dt = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=UTC)
        end_dt = (
            datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
        ).replace(tzinfo=UTC)
    except ValueError as e:
        raise ValueError(f"Invalid date format. Use 'YYYY-MM-DD': {e}") from e

    logger.info(f"Loading flights between {start_date} and {end_date}")

    if self.campaign_service is None:
        raise RuntimeError("Campaign service not initialized")
    try:
        # Get all flights and filter by date
        all_flights = self.campaign_service.get_all_flights()

        filtered_flights = []
        for flight in all_flights:
            takeoff = flight.get("takeoff_datetime")
            if takeoff is None:
                continue
            if isinstance(takeoff, str):
                takeoff = datetime.fromisoformat(takeoff.replace("Z", "+00:00"))
            if takeoff.tzinfo is None:
                takeoff = takeoff.replace(tzinfo=UTC)

            if start_dt <= takeoff < end_dt:
                if campaign_id is None or flight.get("campaign_id") == campaign_id:
                    filtered_flights.append(flight)

        logger.info(f"Loaded {len(filtered_flights)} flights in date range")
        return filtered_flights
    except Exception as e:
        logger.error(f"Error loading flights by date from database: {e}")
        raise

load_specific_data

load_specific_data(flight_id: str, data_types: list[str] | None = None) -> dict[str, list[str]]

Load specific data types from a flight.

Supported data types depend on the flight structure: - 'drone': Drone raw data (from drone folder) - 'aux': Auxiliary data (from aux folder) - 'proc': Processed data (from proc folder) - 'camera': Camera-specific data - 'gps': GPS-specific data - 'imu': IMU-specific data

Parameters:

Name	Type	Description	Default
flight_id	str	Flight ID to load data from	required
data_types	Optional[list[str]]	list of data types to load. If None, loads all available.	None

Returns:

Type	Description
dict[str, list[str]]	dictionary mapping data_type to list of file paths.

Source code in pils/loader/stout.py

def load_specific_data(
    self, flight_id: str, data_types: list[str] | None = None
) -> dict[str, list[str]]:
    """
    Load specific data types from a flight.

    Supported data types depend on the flight structure:
    - 'drone': Drone raw data (from drone folder)
    - 'aux': Auxiliary data (from aux folder)
    - 'proc': Processed data (from proc folder)
    - 'camera': Camera-specific data
    - 'gps': GPS-specific data
    - 'imu': IMU-specific data

    Parameters
    ----------
    flight_id : str
        Flight ID to load data from
    data_types : Optional[list[str]]
        list of data types to load. If None, loads all available.

    Returns
    -------
    dict[str, list[str]]
        dictionary mapping data_type to list of file paths.
    """
    if not flight_id:
        raise ValueError("flight_id is required")

    logger.info(f"Loading specific data for flight {flight_id}: {data_types}")

    # Get flight metadata first
    flight = self.load_single_flight(flight_id=flight_id)
    if not flight:
        raise ValueError(f"Flight {flight_id} not found")

    return self._collect_specific_data(flight, data_types)

get_campaign_list

get_campaign_list() -> list[dict[str, Any]]

Get list of all campaigns.

Returns:

Type	Description
list[dict[str, Any]]	list of campaign dictionaries with metadata.

Source code in pils/loader/stout.py

def get_campaign_list(self) -> list[dict[str, Any]]:
    """
    Get list of all campaigns.

    Returns
    -------
    list[dict[str, Any]]
        list of campaign dictionaries with metadata.
    """
    if self.use_stout and self.campaign_service:
        try:
            return self.campaign_service.get_all_campaigns()
        except Exception as e:
            logger.error(f"Error loading campaigns from database: {e}")
            raise
    else:
        return self._get_campaigns_from_filesystem()

load_flight_data

load_flight_data(flight_id: str | None = None, flight_name: str | None = None, sensors: list[str] | None = None, drones: list[str] | None = None, freq_interpolation: float | None = None, dji_drone_type: str | None = None, drone_correct_timestamp: bool | None = True, polars_interpolation: bool | None = True, align_drone: bool | None = True) -> dict[str, Any]

Load flight data and return dataframes for requested sensors and drones.

This is the main method to load flight data. It returns a dictionary containing flight metadata and dataframes for the requested data types.

Supported sensors: - 'gps': GPS data from UBX/BIN file - 'imu': IMU data - 'adc': ADC sensor data - 'camera': Camera data - 'inclinometer': Inclinometer data

Supported drones: - 'dji': DJI drone telemetry from DAT file - 'litchi': Litchi flight logs - 'blacksquare': BlackSquare drone logs

Parameters:

Name	Type	Description	Default
flight_id	Optional[str]	Flight ID to load	None
flight_name	Optional[str]	Flight name to load (alternative to flight_id)	None
sensors	Optional[list[str]]	list of sensor types to load. If None, loads ['gps'].	None
drones	Optional[list[str]]	list of drone types to load. If None, loads ['dji'].	None

Returns:

Type	Description
dict[str, Any]	dictionary containing: - 'flight_info': Flight metadata dictionary - '': DataFrame for each requested sensor - '': DataFrame for each requested drone

Examples:

>>> loader = StoutDataLoader()
>>> data = loader.load_flight_data(
...     flight_id='some-id',
...     sensors=['gps', 'imu'],
...     drones=['dji']
... )
>>> gps_df = data['gps']
>>> drone_df = data['dji']

Source code in pils/loader/stout.py

def load_flight_data(
    self,
    flight_id: str | None = None,
    flight_name: str | None = None,
    sensors: list[str] | None = None,
    drones: list[str] | None = None,
    freq_interpolation: float | None = None,
    dji_drone_type: str | None = None,
    drone_correct_timestamp: bool | None = True,
    polars_interpolation: bool | None = True,
    align_drone: bool | None = True,
) -> dict[str, Any]:
    """
    Load flight data and return dataframes for requested sensors and drones.

    This is the main method to load flight data. It returns a dictionary
    containing flight metadata and dataframes for the requested data types.

    Supported sensors:
    - 'gps': GPS data from UBX/BIN file
    - 'imu': IMU data
    - 'adc': ADC sensor data
    - 'camera': Camera data
    - 'inclinometer': Inclinometer data

    Supported drones:
    - 'dji': DJI drone telemetry from DAT file
    - 'litchi': Litchi flight logs
    - 'blacksquare': BlackSquare drone logs

    Parameters
    ----------
    flight_id : Optional[str]
        Flight ID to load
    flight_name : Optional[str]
        Flight name to load (alternative to flight_id)
    sensors : Optional[list[str]]
        list of sensor types to load. If None, loads ['gps'].
    drones : Optional[list[str]]
        list of drone types to load. If None, loads ['dji'].

    Returns
    -------
    dict[str, Any]
        dictionary containing:
            - 'flight_info': Flight metadata dictionary
            - '<sensor_type>': DataFrame for each requested sensor
            - '<drone_type>': DataFrame for each requested drone

    Examples
    --------
    >>> loader = StoutDataLoader()
    >>> data = loader.load_flight_data(
    ...     flight_id='some-id',
    ...     sensors=['gps', 'imu'],
    ...     drones=['dji']
    ... )
    >>> gps_df = data['gps']
    >>> drone_df = data['dji']
    """
    if sensors is None:
        sensors = ["gps"]
    if drones is None:
        drones = ["dji"]

    # Load flight metadata
    flight_info = self.load_single_flight(
        flight_id=flight_id, flight_name=flight_name
    )
    if not flight_info:
        raise ValueError(
            f"Flight not found: flight_id={flight_id}, flight_name={flight_name}"
        )

    result: dict[str, Any] = {"flight_info": flight_info}

    # Load requested sensors
    for sensor_type in sensors:
        if sensor_type not in SENSOR_MAP:
            logger.warning(
                f"Unknown sensor type: {sensor_type}. Available: {list(SENSOR_MAP.keys())}"
            )
            continue
        df = self._load_sensor_dataframe(
            flight_info, sensor_type, freq_interpolation
        )
        result[sensor_type] = df
        logger.info(
            f"Loaded {sensor_type} data: {df.shape if df is not None else 'None'}"
        )

    # Load requested drones
    for drone_type in drones:
        if drone_type not in DRONE_MAP:
            logger.warning(
                f"Unknown drone type: {drone_type}. Available: {list(DRONE_MAP.keys())}"
            )
            continue
        if drone_type == "dji" and dji_drone_type is not None:
            dji_drone_type = dji_drone_type
        df = self._load_drone_dataframe(
            flight_info,
            drone_type,
            dji_drone_type,
            drone_correct_timestamp,
            polars_interpolation,
            align_drone,
        )
        result[drone_type] = df
        logger.info(
            f"Loaded {drone_type} data: {'OK' if df is not None else 'None'}"
        )

    return result

get_available_sensors

get_available_sensors() -> list[str]

Return list of available sensor types.

Source code in pils/loader/stout.py

def get_available_sensors(self) -> list[str]:
    """Return list of available sensor types."""
    return list(SENSOR_MAP.keys())

get_available_drones

get_available_drones() -> list[str]

Return list of available drone types.

Source code in pils/loader/stout.py

def get_available_drones(self) -> list[str]:
    """Return list of available drone types."""
    return list(DRONE_MAP.keys())