Data Processing Module Documentation

Overview

The Data Processing module provides utility functions for spatial data processing, file management, and data transformation in the 3-30-300 analysis framework. This module handles tile name translation, spatial filtering, geometry operations, and file format conversions.

Module Information

`filter_buffer_geometries(sedona, geo_level, geo_code, table_name, buffer=None)`

Filters the buffer geometries for a given geo_code.

Parameters:

Name	Type	Description	Default
`sedona`	`SparkSession`	The Spark session.	required
`geo_level`	`str`	The geo_level.	required
`geo_code`	`str`	The geo_code.	required
`table_name`	`str`	The table name.	required
`buffer`	`int`	The buffer size.	`None`

Returns:

Name	Type	Description
`DataFrame`	`DataFrame`	The filtered buffer geometries dataframe.

Source code in src/utils/data_processing.py

def filter_buffer_geometries(sedona: SparkSession, geo_level: str, geo_code: str, table_name: str, buffer: int=None) -> DataFrame:
    """
    Filters the buffer geometries for a given geo_code.

    Args:
        sedona (SparkSession): The Spark session.
        geo_level (str): The geo_level.
        geo_code (str): The geo_code.
        table_name (str): The table name.
        buffer (int): The buffer size.

    Returns:
        DataFrame: The filtered buffer geometries dataframe.
    """

    logging.debug(f"Filtering {table_name} for {geo_code}")

    geo_buildings_sdf = sedona.sql(
        f"""
            SELECT b.* FROM {table_name} b, geo_boundary g 
            WHERE ST_Intersects(b.geometry, g.geometry)
        """)
    geo_buildings_sdf.createOrReplaceTempView(f"geo_{table_name}")

    if buffer:

        geo_buildings_buffer_sdf = sedona.sql(
        f"""
            SELECT ST_Buffer(b.geometry, {buffer}) AS geometry, b.verisk_premise_id
            FROM geo_{table_name} b
        """)
        geo_buildings_buffer_sdf.createOrReplaceTempView(f"{table_name}_buffers")

        return geo_buildings_buffer_sdf

    return geo_buildings_sdf

`generate_tile_paths(geo_level, geo_code, output_areas_os_tile_overlay_df, vom_raster_paths_df, tree_vector_paths_df)`

Generates the tile paths for a given geo_code.

Parameters:

Name	Type	Description	Default
`geo_level`	`str`	The geo_level.	required
`geo_code`	`str`	The geo_code.	required
`output_areas_os_tile_overlay_df`	`GeoDataFrame`	The output areas OS tile overlay dataframe.	required
`vom_raster_paths_df`	`DataFrame`	The VOM raster paths dataframe.	required
`tree_vector_paths_df`	`DataFrame`	The tree vector paths dataframe.	required

Returns:

Type	Description
`DataFrame`	pd.DataFrame: The tile paths dataframe.

Source code in src/utils/data_processing.py

def generate_tile_paths(geo_level: str, geo_code: str, output_areas_os_tile_overlay_df: gpd.GeoDataFrame, vom_raster_paths_df: pd.DataFrame, tree_vector_paths_df: pd.DataFrame) -> pd.DataFrame:
    """
    Generates the tile paths for a given geo_code.

    Args:
        geo_level (str): The geo_level.
        geo_code (str): The geo_code.
        output_areas_os_tile_overlay_df (gpd.GeoDataFrame): The output areas OS tile overlay dataframe.
        vom_raster_paths_df (pd.DataFrame): The VOM raster paths dataframe.
        tree_vector_paths_df (pd.DataFrame): The tree vector paths dataframe.

    Returns:
        pd.DataFrame: The tile paths dataframe.
    """

    logging.debug(f"Generating tile (vom and tree) paths for {geo_code}")

    geo_output_areas_os_tile_overlay_df = output_areas_os_tile_overlay_df.copy()[output_areas_os_tile_overlay_df[geo_level] == geo_code]
    vom_raster_paths_df['TILE_NAME_5KM_int'] = vom_raster_paths_df.TILE_NAME.apply(lambda x: x.lower())
    tree_vector_paths_df['TILE_NAME_5KM_int'] = tree_vector_paths_df.TILE_NAME.apply(lambda x: x.lower())
    geo_vom_tiles_df = geo_output_areas_os_tile_overlay_df.merge(vom_raster_paths_df, on='TILE_NAME_5KM_int', how='left')[['TILE_NAME', 'year', 'path']].drop_duplicates().sort_values(['TILE_NAME', 'year'], ascending=[True, False]).reset_index(drop=True)
    geo_tree_tiles_df = geo_output_areas_os_tile_overlay_df.merge(tree_vector_paths_df, on='TILE_NAME_5KM_int', how='left')[['TILE_NAME', 'year', 'path']].drop_duplicates().sort_values(['TILE_NAME', 'year'], ascending=[True, False]).reset_index(drop=True)

    geo_tiles_df = geo_vom_tiles_df.merge(geo_tree_tiles_df, on=['TILE_NAME', 'year'], suffixes=['_vom', '_tree'])

    return geo_tiles_df

`get_geometries(sedona, geo_level, geo_code, dissolve=True)`

Gets the geometries for a given geo_code.

Parameters:

Name	Type	Description	Default
`sedona`	`SparkSession`	The Spark session.	required
`geo_level`	`str`	The geo_level.	required
`geo_code`	`str`	The geo_code.	required
`dissolve`	`bool`	Whether to dissolve the geometries.	`True`

Returns:

Name	Type	Description
`DataFrame`	`DataFrame`	The geometries dataframe.

Source code in src/utils/data_processing.py

def get_geometries(sedona: SparkSession, geo_level: str, geo_code: str, dissolve: bool=True) -> DataFrame:
    """
    Gets the geometries for a given geo_code.

    Args:
        sedona (SparkSession): The Spark session.
        geo_level (str): The geo_level.
        geo_code (str): The geo_code.
        dissolve (bool): Whether to dissolve the geometries.

    Returns:
        DataFrame: The geometries dataframe.
    """

    logging.debug(f"Dissolving geometries for {geo_level}:{geo_code}")

    query = "ST_Union_Aggr(geometry) AS geometry" if dissolve else "*"

    geo_boundary_sdf = sedona.sql(
        f"""
            SELECT {query}
            FROM boundaries
            WHERE {geo_level} = '{geo_code}'
        """)
    geo_boundary_sdf.createOrReplaceTempView("geo_boundary")

    return geo_boundary_sdf

`get_overlapping_grid_tiles(output_areas_boundaries_gdf, os_tile_boundaries_gdf, geo_level, geo_code, tile_level)`

Gets the overlapping grid tiles for a given geo_code and tile_level.

Parameters:

Name	Type	Description	Default
`output_areas_boundaries_gdf`	`GeoDataFrame`	The output areas boundaries dataframe.	required
`os_tile_boundaries_gdf`	`GeoDataFrame`	The OS tile boundaries dataframe.	required
`geo_level`	`str`	The geo_level.	required
`geo_code`	`str`	The geo_code.	required
`tile_level`	`str`	The tile_level.	required

Returns:

Name	Type	Description
`list`	`list`	The overlapping grid tiles.

Source code in src/utils/data_processing.py

def get_overlapping_grid_tiles(output_areas_boundaries_gdf: gpd.GeoDataFrame, os_tile_boundaries_gdf: gpd.GeoDataFrame, geo_level: str, geo_code: str, tile_level: str) -> list:
    """
    Gets the overlapping grid tiles for a given geo_code and tile_level.

    Args:
        output_areas_boundaries_gdf (gpd.GeoDataFrame): The output areas boundaries dataframe.
        os_tile_boundaries_gdf (gpd.GeoDataFrame): The OS tile boundaries dataframe.
        geo_level (str): The geo_level.
        geo_code (str): The geo_code.
        tile_level (str): The tile_level.

    Returns:
        list: The overlapping grid tiles.
    """

    logging.debug(f"Getting overlapping grid tiles for {geo_code} and {tile_level}")

    selected_feature = output_areas_boundaries_gdf[output_areas_boundaries_gdf[geo_level] == geo_code]

    # Get the overlapping features
    overlapping_tiles_lst = gpd.overlay(selected_feature, os_tile_boundaries_gdf, how='intersection')[tile_level].unique().tolist()

    return overlapping_tiles_lst

`save_csv_as_parquet(in_directory, path_pattern, out_path)`

Saves the CSV files as a parquet file.

Parameters:

Name	Type	Description	Default
`in_directory`	`Path`	The input directory.	required
`path_pattern`	`str`	The path pattern.	required
`out_path`	`Path`	The output path.	required

Returns:

Type	Description
`DataFrame`	pd.DataFrame: The concatenated dataframe.

Source code in src/utils/data_processing.py

def save_csv_as_parquet(in_directory: Path, path_pattern: str, out_path: Path) -> pd.DataFrame:
    """
    Saves the CSV files as a parquet file.

    Args:
        in_directory (Path): The input directory.
        path_pattern (str): The path pattern.
        out_path (Path): The output path.

    Returns:
        pd.DataFrame: The concatenated dataframe.
    """

    csv_files = list(in_directory.glob(path_pattern))
    dataframes_lst = [pd.read_csv(file) for file in csv_files]
    concatenated_df = pd.concat(dataframes_lst, ignore_index=True)
    concatenated_df.to_parquet(out_path, index=False)

    return concatenated_df

`save_temp_file(spark_df, output_path, coalesce=1, file_format='csv')`

Saves a Spark DataFrame to a single named file and returns a Pandas DataFrame.

This function is a workaround for the fact that Spark normally writes to a directory. It works by coalescing the DataFrame to a single partition, writing to a temporary directory, and then moving the single generated data file to the final destination.

Parameters:

Name	Type	Description	Default
`spark_df`	`DataFrame`	The Spark DataFrame to save.	required
`output_path`	`Path`	The final, full path for the output file (e.g., Path("/data/my_file.parquet")).	required
`file_format`	`str`	The format to save the file in ("parquet", "csv", etc.).	`'csv'`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: The saved data loaded into a Pandas DataFrame.

Source code in src/utils/data_processing.py

def save_temp_file(spark_df: DataFrame, output_path: Path, coalesce: int=1, file_format: str="csv") -> pd.DataFrame:
    """
    Saves a Spark DataFrame to a single named file and returns a Pandas DataFrame.

    This function is a workaround for the fact that Spark normally writes to a directory.
    It works by coalescing the DataFrame to a single partition, writing to a temporary
    directory, and then moving the single generated data file to the final destination.

    Args:
        spark_df (DataFrame): The Spark DataFrame to save.
        output_path (Path): The final, full path for the output file (e.g., Path("/data/my_file.parquet")).
        file_format (str): The format to save the file in ("parquet", "csv", etc.).

    Returns:
        pd.DataFrame: The saved data loaded into a Pandas DataFrame.
    """

    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)

        # Step 1: Write the coalesced DataFrame to the temporary directory.
        # Spark will create a directory with part-files and metadata inside.
        spark_df.coalesce(coalesce) \
            .write \
            .option("header", "true") \
            .mode("overwrite") \
            .format(file_format) \
            .save(str(temp_path))

        # Step 2: Find the single part-file Spark wrote.
        # It will have a name like 'part-00000-....c000.snappy.parquet'
        part_files = list(temp_path.glob(f"part-*.{file_format}*"))
        if not part_files:
            raise FileNotFoundError(f"No part file found with format '{file_format}' in {temp_dir}")

        temp_file = part_files[0]

        # Step 3: Move and rename the part-file to your target path.
        # This moves the actual data file to its final destination.
        shutil.move(temp_file, output_path)

    # Step 4: Read the final file into Pandas using the correct reader.
    if file_format == "parquet":
        pandas_df = pd.read_parquet(output_path)
    elif file_format == "csv":
        pandas_df = pd.read_csv(output_path)
    else:
        # Add other formats as needed or raise an error
        raise ValueError(f"Unsupported file format for reading into Pandas: {file_format}")

    return pandas_df

`translate_tile_name(tile_name)`

Translates a tile name between two formats:

Format 1: TL0045 (where '0045' represents coordinates)
Format 2: TL04NW (where 'NW' represents directions)

The function converts: - From Format 1 to Format 2 by interpreting the numeric coordinates and converting them to directional codes. - From Format 2 to Format 1 by interpreting the directional codes and converting them to numeric coordinates.

Parameters:

Name	Type	Description	Default
`tile_name`	`str`	The tile name to be translated. It should be a string of length 6.	required

Returns:

Name	Type	Description
`str`	`str`	The translated tile name in the opposite format.

Raises:

Type	Description
`AssertionError`	If the input tile_name is not of length 6.
`ValueError`	If the numeric part of the tile name cannot be converted to an integer when expected.

Source code in src/utils/data_processing.py

def translate_tile_name(tile_name: str) -> str:
    """
    Translates a tile name between two formats:

    - Format 1: TL0045 (where '0045' represents coordinates)
    - Format 2: TL04NW (where 'NW' represents directions)

    The function converts:
    - From Format 1 to Format 2 by interpreting the numeric coordinates and converting them to directional codes.
    - From Format 2 to Format 1 by interpreting the directional codes and converting them to numeric coordinates.

    Args:
        tile_name (str): The tile name to be translated. It should be a string of length 6.

    Returns:
        str: The translated tile name in the opposite format.

    Raises:
        AssertionError: If the input tile_name is not of length 6.
        ValueError: If the numeric part of the tile name cannot be converted to an integer when expected.
    """

    NS_dict = {'S': '0', 'N': '5'}
    EW_dict = {'W': '0', 'E': '5'} 

    assert len(tile_name) == 6

    code = tile_name[2:6].upper()
    try: # If input is like TL0045
        int(code)
        NS_dict = {v: k for k, v in NS_dict.items()}
        EW_dict = {v: k for k, v in EW_dict.items()}
        ns_id = code[3]
        ew_id = code[1]
        direction_code = code[0] + code[2] + NS_dict[ns_id] + EW_dict[ew_id]
        trans_tile_name = tile_name[:2].upper() + direction_code
    except ValueError: # If input is like TL04NW
        ns_id = code[2]
        ew_id = code[3]
        number_code = code[0] + EW_dict[ew_id] + code[1] + NS_dict[ns_id]
        trans_tile_name = tile_name[:2].lower() + number_code

    return trans_tile_name

Usage Examples

Basic Workflow

The typical workflow for data processing involves:

Tile name translation - Converting between different tile naming formats
Spatial filtering - Extracting geometries for specific geographic areas
Path generation - Creating file paths for VOM and tree data
File conversion - Converting between different file formats

Example: Translating Tile Names

from src.utils.data_processing import translate_tile_name

# Convert from numeric to directional format
numeric_tile = "TL0045"
directional_tile = translate_tile_name(numeric_tile)
print(f"{numeric_tile} -> {directional_tile}")  # TL0045 -> TL04NW

# Convert from directional to numeric format
directional_tile = "TL04NW"
numeric_tile = translate_tile_name(directional_tile)
print(f"{directional_tile} -> {numeric_tile}")  # TL04NW -> tl0045

Example: Getting Overlapping Grid Tiles

from src.utils.data_processing import get_overlapping_grid_tiles

# Get overlapping tiles for a geographic area
overlapping_tiles = get_overlapping_grid_tiles(
    output_areas_boundaries_gdf=boundaries_gdf,
    os_tile_boundaries_gdf=tiles_gdf,
    geo_level="LAD22CD",
    geo_code="E06000001",
    tile_level="TILE_NAME_5KM"
)

Example: Generating Tile Paths

from src.utils.data_processing import generate_tile_paths

# Generate paths for VOM and tree data
tile_paths_df = generate_tile_paths(
    geo_level="LAD22CD",
    geo_code="E06000001",
    output_areas_os_tile_overlay_df=overlay_df,
    vom_raster_paths_df=vom_paths_df,
    tree_vector_paths_df=tree_paths_df
)

Example: Filtering Buffer Geometries

from src.utils.data_processing import filter_buffer_geometries

# Filter buildings with buffer
buildings_buffer_sdf = filter_buffer_geometries(
    sedona=sedona,
    geo_level="LAD22CD",
    geo_code="E06000001",
    table_name="buildings",
    buffer=100
)

Example: Getting Geometries

from src.utils.data_processing import get_geometries

# Get dissolved geometries for an area
boundary_sdf = get_geometries(
    sedona=sedona,
    geo_level="LAD22CD",
    geo_code="E06000001",
    dissolve=True
)

Example: Saving Files

from src.utils.data_processing import save_temp_file, save_csv_as_parquet
from pathlib import Path

# Save Spark DataFrame as single file
output_path = Path("output/data.parquet")
pandas_df = save_temp_file(
    spark_df=result_sdf,
    output_path=output_path,
    coalesce=1,
    file_format="parquet"
)

# Convert CSV files to parquet
csv_df = save_csv_as_parquet(
    in_directory=Path("input/"),
    path_pattern="*.csv",
    out_path=Path("output/combined.parquet")
)

Tile Name Translation

Tile Name Formats

Format 1: Numeric Coordinates

Pattern: TL0045 (where '0045' represents coordinates)
Structure: 2-letter prefix + 4-digit numeric code
Example: TL0045, SU1234, SP5678

Format 2: Directional Codes

Pattern: TL04NW (where 'NW' represents directions)
Structure: 2-letter prefix + 2-digit number + 2-letter direction
Example: TL04NW, SU12SE, SP56NE

Translation Logic

Numeric to Directional: Converts coordinate numbers to directional codes
Directional to Numeric: Converts directional codes to coordinate numbers
Case Handling: Maintains appropriate case for each format
Validation: Ensures input tile names are 6 characters long

Spatial Operations

Grid Tile Overlap

Spatial Intersection: Finds tiles that overlap with geographic areas
Efficient Filtering: Uses spatial indexing for performance
Result Aggregation: Returns unique tile names for processing
Coordinate System: Works with project coordinate reference system

Geometry Filtering

Spatial Joins: Uses ST_Intersects for spatial relationships
Buffer Operations: Creates spatial buffers around geometries
Dissolution: Combines multiple geometries into single boundary
Temporary Views: Creates Spark SQL views for efficient querying

File Management

Path Generation

Tile Matching: Matches geographic areas with available data tiles
Year Sorting: Orders data by year (newest first)
Duplicate Removal: Eliminates duplicate tile-year combinations
Path Validation: Ensures file paths exist and are accessible

File Format Conversion

CSV to Parquet: Converts CSV files to efficient parquet format
Spark to Pandas: Converts Spark DataFrames to Pandas DataFrames
Single File Output: Ensures single output file instead of directory
Format Validation: Supports multiple output formats

Key Parameters

Tile Translation Parameters

tile_name: 6-character tile name to translate
format: Source format (numeric or directional)
case: Output case (upper or lower)

Spatial Operation Parameters

geo_level: Geographic level (LAD22CD, MSOA21CD, etc.)
geo_code: Specific geographic code
buffer: Buffer distance in meters
dissolve: Whether to dissolve geometries

File Operation Parameters

output_path: Target file path
file_format: Output format (parquet, csv)
coalesce: Number of partitions for output
path_pattern: File pattern for glob matching

Data Flow

Input Data

Geographic Boundaries: Output area boundaries and tile boundaries
VOM Data: Raster file paths and metadata
Tree Data: Vector file paths and metadata
Building Data: Building geometries and attributes

Processing Steps

Tile Name Translation: Converts between naming formats
Spatial Filtering: Extracts relevant geographic areas
Path Generation: Creates file paths for data access
Geometry Operations: Performs spatial analysis
File Conversion: Converts between file formats

Output Data

Translated Names: Tile names in target format
Filtered Geometries: Spatial data for specific areas
File Paths: Organized paths for data access
Converted Files: Data in target format

Performance Optimizations

Spatial Operations

Spatial Indexing: Uses spatial indices for efficient queries
Partitioning: Distributes processing across cluster nodes
Caching: Caches frequently accessed geometries
Batch Processing: Processes multiple areas efficiently

File Operations

Coalescing: Reduces number of output files
Compression: Uses efficient file compression
Parallel Processing: Processes files in parallel
Memory Management: Optimizes memory usage for large files

Error Handling

The module includes comprehensive error handling for:

Invalid tile names: Validates tile name format and length
Missing files: Handles missing data files gracefully
Spatial errors: Manages spatial operation failures
File system issues: Handles file permission and path issues
Memory errors: Manages large dataset memory requirements

Dependencies

This module requires:

pandas for data manipulation
geopandas for spatial data handling
pyspark for distributed processing
pathlib for cross-platform path handling
tempfile for temporary file management
shutil for file operations

Notes

The module provides standardized data processing utilities
All spatial operations use the project's coordinate reference system
Tile name translation supports OS (Ordnance Survey) conventions
File operations are optimized for large-scale processing
The module includes comprehensive error handling and validation
Supports both local and distributed processing scenarios
Maintains data integrity throughout processing pipeline