checkers.py

checker functions check a file against the rules. They also catch errors to do with opening the files.

check_bgs(filename, **kwargs)

Validate file against BGS rules. kwargs parameter required because some validation functions have keyword parameters.

Source code in app/checkers.py

def check_bgs(filename: Path, **kwargs) -> dict:
    """
    Validate file against BGS rules.  kwargs parameter required because some
    validation functions have keyword parameters.
    """
    logger.info("Checking %s against BGS rules.", filename.name)
    errors = {}
    load_error = None
    bgs_metadata = {}

    tables, load_error, ags4_errors = load_tables_reporting_errors(filename)

    if load_error:
        errors["File read error"] = [{"line": "-", "group": "", "desc": load_error}]
        error_count = 1
    else:
        errors.update(ags4_errors)
        # Get additional metadata
        bgs_metadata = generate_bgs_metadata(tables)

        # Apply checks
        error_count = 0
        for rule, func in BGS_RULES.items():
            logger.info("Checking against %s", rule)
            result = func(tables)
            if result:
                errors[rule] = result
                error_count += 1

    return dict(
        checker=f"bgs_rules v{bgs_rules_version}",
        errors=errors,
        error_count=error_count,
        warnings_count=0,
        fyi_count=0,
        additional_metadata=bgs_metadata,
    )

generate_bgs_metadata(tables)

Generate additional metadata from groups.

Source code in app/checkers.py

def generate_bgs_metadata(tables: Dict[str, pd.DataFrame]) -> dict:
    """Generate additional metadata from groups."""
    try:
        projects = (
            tables["PROJ"]
            .apply(lambda row: f"{row['PROJ_ID']} ({row['PROJ_NAME']})", axis=1)
            .to_list()
        )
    except KeyError:
        projects = []

    try:
        loca_rows = len(tables["LOCA"][tables["LOCA"]["HEADING"] == "DATA"])
    except KeyError:
        loca_rows = 0

    groups = tables.keys()
    bgs_metadata = {
        "bgs_all_groups": f"{len(groups)} groups identified in file: {' '.join(groups)}",
        "bgs_file": f"Optional FILE group present: {'FILE' in groups}",
        "bgs_dict": f"Optional DICT group present: {'DICT' in groups}",
        "bgs_loca_rows": f"{loca_rows} data row(s) in LOCA group",
        "bgs_projects": f"{len(projects)} projects found: {'; '.join(projects)}",
    }
    return bgs_metadata

load_ags4_as_numeric(filename)

Read AGS4 file and convert to numeric data types.

Source code in app/checkers.py

def load_ags4_as_numeric(filename: Path) -> Tuple[dict, dict, List[dict]]:
    """Read AGS4 file and convert to numeric data types."""
    tables, headings = AGS4.AGS4_to_dataframe(filename)

    # Check the TYPE of coordinate in LOCA
    coord_columns = ["LOCA_NATE", "LOCA_NATN", "LOCA_LOCX", "LOCA_LOCY"]
    errors = get_coord_column_type_errors(tables, coord_columns)

    # Convert tables to numeric data for analysis
    for group, df in tables.items():
        tables[group] = AGS4.convert_to_numeric(df)

    # Force conversion of LOCA coordinate columns, even if type is not numeric
    if tables:
        for column in coord_columns:
            try:
                tables["LOCA"][column] = pd.to_numeric(tables["LOCA"][column])
            except KeyError:
                # Not all files have all columns
                pass

    return tables, headings, errors

get_coord_column_type_errors(tables, coord_columns)

Check the coordinate columns in LOCA table have correct data type and return errors for those that don't.

Source code in app/checkers.py

def get_coord_column_type_errors(tables: dict, coord_columns: List[str]) -> dict:
    """
    Check the coordinate columns in LOCA table have correct data type and
    return errors for those that don't.
    """
    try:
        loca = tables["LOCA"]
    except KeyError:
        # If LOCA doesn't exist, other errors are returned elsewhere
        return {}

    bad_columns = []
    for column in coord_columns:
        try:
            type_ = loca.loc[loca["HEADING"] == "TYPE", column].tolist()[0]
            if not re.search(r"(DP|MC|SF|SCI)", type_):
                bad_columns.append(f"{column} ({type_})")
        except KeyError:
            # Ignore columns that don't exist
            pass

    if bad_columns:
        error_message = (
            f"Coordinate columns have non-numeric TYPE: {', '.join(bad_columns)}"
        )
        errors = {
            "BGS data validation: Non-numeric coordinate types": [
                {"line": "-", "group": "LOCA", "desc": error_message}
            ]
        }
    else:
        errors = {}

    return errors