Utils Reference¶

def find_address_fields(config) -> dict[str]:
    """
    Parses which address fields to consider in the input file based on
    the content of config.yml. Raises an error if neither full_address_field
    nor street are specified in the config file.

    Args:
        config (dict): A config object

    Returns dict: A dict of address field names in the input file.

    """
    # There are two possible ways to input address in the yaml config
    # 1. Specifying a full address string (if address is stored in one column)
    # 2. Specifying a list of address fields (address, city, state, zip) for
    # if address is stored in multiple columns.
    full_addr = config.get("full_address_field")

    addr_fields = config.get("address_fields") or {}

    # street_address used to be called street, adding this for backward compatibility
    # in case someone hasn't updated their config file to call it street_address
    if addr_fields.get("street") and not addr_fields.get("street_address"):
        addr_fields["street_address"] = addr_fields.pop("street")

    # If user has not specified an address field, raise
    if not full_addr and not any(addr_fields.values()):
        raise ValueError(
            "An address field or address fields must be specified in the config file."
        )

    # Handle cases where user has specified both a full address field
    # and separate address fields.
    resp = ""

    if full_addr and addr_fields:
        print(
            "You have specified both a full address and separate\n"
            "address fields in the config file.\n"
            "Press 1 to use the full address.\n"
            "Press 2 to use the address fields.\n"
            "Press any other key to quit.\n"
        )

        while resp.lower() not in ["1", "2", "q", "quit"]:
            if full_addr and addr_fields:
                resp = input("Specify which fields to use: ")

            if resp == "1":
                return {"full_address": full_addr}

            elif resp == "2":
                break

            else:
                print("Exiting program...")
                sys.exit()

    if full_addr and not resp:
        return {"full_address": full_addr}

    if not addr_fields.get("street_address"):
        raise ValueError(
            "When full address field is not specified, "
            "address_fields must include a non-null value for "
            "street_address."
        )

    fields = {k: v for k, v in addr_fields.items() if v is not None}
    return fields

def parse_address(parser, address: str) -> tuple[str, bool, bool]:
    """
    Given an address string, uses PassyunkParser to return
    a standardized address, and whether or not the given string
    is an extant address in Philadelphia. Makes some attempt
    to normalize alternate spellings of addresses: eg, 123 Mkt will
    evaluate to 123 MARKET ST

    Args:
        parser: A PassyunkParser object
        address: An address string

    Returns tuple(str, bool, bool): tuple with the standardized address, a
    boolean value indicating if the string is formatted as an address,
    and a boolean value indicating if the address is a valid Philadelphia
    address.
    """

    try:
        prsd = parser.parse(address)
        parsed = prsd["components"]

        has_street_code = False
        for street in ("street", "street_2"):
            sc = parsed.get(street, {}).get("street_code")
            if sc:
                has_street_code = True
                break

        # If address matches to a street code, it is a philly address
        is_addr = bool(has_street_code)
        is_philly_addr = bool(has_street_code)

        output_address = parsed.get("output_address", address)

    # Handle Passyunk parsing edge cases
    except Exception as e:
        output_address = address
        is_addr = False
        is_philly_addr = False

    return {
        "output_address": output_address,
        "is_addr": is_addr,
        "is_philly_addr": is_philly_addr,
        "is_multiple_match": False,
        "geocoder_used": None,
    }

def is_non_philly(address: str | dict | None, address_is_split: bool, zips) -> dict:
    """Determines whether an address is in Philadelphia. Handles
    A string full address or a dict split address.

    Args:
        address (str | dict | None)
        address_is_split (bool): whether or not the address is the full address, or split
        zips: list of Philadelphia zip codes

    Returns:
        dict: {'is_non_philly': bool, 'is_undefined': bool}
    """
    if not address:
        return {"is_non_philly": False, "is_undefined": True}

    # If address is in full address form, we need to tag it
    address_data = address if address_is_split else tag_full_address(address)

    return flag_non_philly_address(address_data, zips)

@retry(
    wait_exponential_multiplier=1000,
    wait_exponential_max=10000,
    stop_max_attempt_number=3,
    wait_fixed=200,
)
def ais_lookup(
    sess: requests.Session,
    api_key: str,
    address: str,
    zip: str = None,
    enrichment_fields: list = [],
    existing_is_addr: bool = False,
    existing_is_philly_addr: bool = False,
    original_address: str = None,
    fetch_4326: bool = True,
    fetch_2272: bool = True,
) -> dict:
    """
    Given a passyunk-normalized address, looks up whether or not it is in the
    database.

    Args:
        sess (requests Session object): A requests library session object
        api_key (str): An AIS api key
        address (str): The address to query
        zip (str): The zip code associated with the address, if present
        enrichment_fields (list): The fields to add from AIS
        fetch_4326 (bool): Whether to fetch SRID 4326 coordinates (lat/lon)
        fetch_2272 (bool): Whether to fetch SRID 2272 coordinates (x/y)

    Returns:
        A dict with standardized address, latitude and longitude,
        and user-requested fields.
    """
    AIS_RATE_LIMITER.wait()

    # Don't attempt to geocode if address is null
    if address:
        ais_url = f"https://api-prod.phila.gov/ais/v1/search/{quote(address)}"

        params = {}

        # To handle backwards compatibility with people still using a gatekeeper key instead of
        # a client ID, we set both params here
        params["gatekeeperKey"] = api_key
        params["client_id"] = api_key
        params["srid"] = 4326
        params["max_range"] = 0

        try:
            response = sess.get(ais_url, params=params)
        except:
            print(
                f"Warning: AIS lookup failed for this address: {address}, {zip}, {original_address}"
            )
            response = None
    else:
        response = None

    if response and response.status_code >= 500:
        raise Exception("5xx response. There may be a problem with the AIS API.")
    elif response and response.status_code == 429:
        raise Exception("429 response. Too many calls to the AIS API.")

    # Initialize lat and lon values
    (
        lat,
        lon,
    ) = (
        None,
        None,
    )
    geocode_lat, geocode_lon, geocode_x, geocode_y = None, None, None, None

    # If status code is 200, that means API has found a match.
    # API will return a 404 if no match
    if response and response.status_code == 200:
        # If r_json is longer than 1, multiple matches
        # were returned and we need to tiebreak
        r_json = response.json()

        search_type = r_json.get("search_type")

        # Intersection returns a different data structure with fewer
        # possible enrichment fields, so we need to handle this differently
        if search_type == "intersection":

            parsed_response = parse_intersection_lookup(
                sess, api_key, r_json, original_address, zip, enrichment_fields
            )

            # If tiebreak fails, return
            # null values for most fields.
            if not parsed_response:
                ais_result = AISResult(
                    output_address=original_address if original_address else address,
                    is_addr=False,
                    is_philly_addr=True,
                    is_multiple_match=False,
                    geocoder_used="ais-intersection",
                )

                return asdict(ais_result)

        else:

            parsed_response = parse_address_lookup(r_json, zip, enrichment_fields)

            if not parsed_response:
                # If no match, return
                # null values for most fields.
                # Tiebreaking has failed in this case
                # so is_multiple_match = True
                normalized_addr = r_json.get("normalized", "")

                ais_result = AISResult(
                    output_address=normalized_addr if normalized_addr else address,
                    is_addr=False,
                    is_philly_addr=True,
                    is_multiple_match=True,
                    geocoder_used="ais-full-match",
                )

                return asdict(ais_result)


        # We use the original address here because the address that we use
        # to search against AIS may be augmented with PHILADELPHIA, PA
        # if no city, state exists

        lat = parsed_response["lat"]
        lon = parsed_response["lon"]
        out_address = parsed_response["output_address"]

        if fetch_4326:
            # Don't need to make another lookup, we already have
            # coords from first lookup
            # get latitude and longitude from address search only
            # other searches -- against the service_areas endpoint
            # return lat/lon with less precision, so we just use the original
            # lat, lon

            geocode_lat = _round_coordinates(lat)
            geocode_lon = _round_coordinates(lon)

        if fetch_2272:
            geo_x, geo_y = _fetch_ais_coordinates(sess, api_key, out_address, zip, 2272)

            geocode_x = _round_coordinates(geo_x)
            geocode_y = _round_coordinates(geo_y)

        ais_result = AISResult(
            output_address=out_address if out_address else address,
            is_addr=True if search_type == "address" else False,
            is_philly_addr=True,
            is_multiple_match=False,
            geocoder_used=(
                "ais-full-match" if search_type == "address" else "ais-intersection"
            ),
            geocode_lat=geocode_lat,
            geocode_lon=geocode_lon,
            geocode_x=geocode_x,
            geocode_y=geocode_y,
        )

        return asdict(ais_result) | parsed_response["enriched_fields"]

    # If no match, return none but preserve existing address validity flags
    # Use original_address if provided, otherwise fall back to address parameter
    ais_result = AISResult(
        output_address=original_address if original_address else address,
        is_addr=existing_is_addr,
        is_philly_addr=existing_is_philly_addr,
        is_multiple_match=False,
    )

    return asdict(ais_result)

def fetch_service_area_enrichment_data(
    sess: requests.Session, api_key: str, lat: str, lon: str, enrichment_fields: list
) -> dict:
    """
    Looks up latitude and longitude against the AIS API service area endpoint,

    Args:
        sess: a requests Session object
        api_key: an AIS API key
        lat: latitude
        lon: longitude
        enrichment_fields: which fields to return from AIS

    Returns:
        (dict): A dictionary of enrichment data
    """

    result = _lookup_service_area(sess, lat, lon, api_key)

    if not result:
        return {}

    return {
        field: result.get("service_areas", {}).get(field) for field in enrichment_fields
    }

@retry(
    wait_exponential_multiplier=1000,
    wait_exponential_max=10000,
    stop_max_attempt_number=5,
)
def tomtom_lookup(
    sess: requests.Session,
    parser,
    api_key: str,
    tomtom_url: str,
    philly_zips: list,
    address: str,
    fallback_addr,
    fetch_4326: bool = True,
    fetch_2272: bool = True,
) -> dict:
    """
    Given a passyunk-normalized address, looks up via TomTom.

    Args:
        sess (requests Session object): A requests library session object
        parser: A passyunk parser object, used to normalize output
        api_key (str): The AIS API key, also used for TomTom
        tomtom_url (str): The TomTom URL to use. 
            Present for backwards compatibility with people using API keys 
            that are not valid for the new Mulesoft gateway endpoint.
        philly_zips (list): A list of philadelphia zips to validate
        tomtom output against
        address (str): The address to query
        fallback_addr (str): The address to return if no match is found
        fetch_4326 (bool): Whether or not to pull coordinates in 4326
        fetch_2272 (bool): Whether or not to pull coordinates in 2272

    Returns:
        A dict with standardized address, latitude and longitude, returned
        from TomTom.
    """
    out_data = _do_tomtom_lookup(
        sess, parser, api_key, tomtom_url, philly_zips, address, fetch_4326, fetch_2272
    )

    if out_data is not None:
        return out_data

    # Truly no match
    out_data = {
        "output_address": fallback_addr if fallback_addr else address,
        "geocoder_used": None,
        "is_addr": False,
        "is_philly_addr": False,
    }

    if fetch_4326:
        out_data["geocode_lat"] = None
        out_data["geocode_lon"] = None

    if fetch_2272:
        out_data["geocode_x"] = None
        out_data["geocode_y"] = None

    return out_data

def detect_file_encoding(file_path: str):
    # Attempt to determine the filetype
    # by reading the first 100kb of a file
    with open(file_path, "rb") as f:
        raw_data = f.read(100000)

    result = chardet.detect(raw_data)
    encoding = result["encoding"]

    # If there's a utf-8 character later in the file
    # the file will be encoded as ascii when in fact
    # it is utf-8. Because utf-8 encompasses ascii,
    # just return encoding as utf-8 in case
    # we can treat ascii as utf-8 (not vice versa)
    if encoding and encoding.lower() == "ascii":
        encoding = "utf-8"

    return encoding

def recode_to_utf8(src_path: str, dst_path: str, src_encoding: str) -> Path:
    """
    Reincode an input file to account for non-standard characters, line by line.
    """

    src = Path(src_path)
    if dst_path is None:
        dst = src.with_suffix(src.suffix + ".utf8")

    else:
        dst = Path(dst_path)

    with (
        src.open("r", encoding=src_encoding, errors="strict", newline="") as fin,
        dst.open("w", encoding="utf-8", newline="") as fout,
    ):
        for line in fin:
            fout.write(line)