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