from config import Config
from util.mongodb import MongoDbUtil
[docs]
class RADUtil:
"""Utilities work working with RAD db."""
[docs]
@staticmethod
def get_naics_codes() -> dict:
"""Get 2022 NAICS codes from US Census. Source: https://www.naics.com/search/
:return: dict of NAICS codes
"""
naics_codes = {
11: "Agriculture, Forestry, Fishing and Hunting",
21: "Mining, Quarrying, and Oil and Gas Extraction",
22: "Utilities",
23: "Construction",
31: "Manufacturing",
32: "Manufacturing",
33: "Manufacturing",
42: "Wholesale Trade",
44: "Retail Trade",
45: "Retail Trade",
48: "Transportation and Warehousing",
49: "Transportation and Warehousing",
51: "Information",
52: "Finance and Insurance",
53: "Real Estate and Rental and Leasing",
54: "Professional, Scientific, and Technical Services",
55: "Management of Companies and Enterprises",
56: "Administrative and Support and Waste Management and Remediation Services",
61: "Educational Services",
62: "Health Care and Social Assistance",
71: "Arts, Entertainment, and Recreation",
72: "Accommodation and Food Services",
81: "Other Services (except Public Administration)",
92: "Public Administration",
}
return naics_codes
[docs]
@staticmethod
def create_or_get_rad_source(source_name: str) -> str:
"""
Create or get source_id from RAD database.
:param source_name: name of the source
:return: source_id
"""
import datetime
source = MongoDbUtil.mongodb_find(
Config.RAD_PREFIXES.DISCOVERY_SOURCES_TABLE,
{"name": source_name},
find_one=True,
)
if source:
return source["_id"]
source = MongoDbUtil.mongodb_insert_one(
Config.RAD_PREFIXES.DISCOVERY_SOURCES_TABLE,
{
"name": source_name,
"created_at": datetime.datetime.now(),
"updated_at": datetime.datetime.now(),
"active": True,
},
)
return str(source.inserted_id)
[docs]
@staticmethod
def get_null_string_uppercased():
"""Get list of common values for null strings. NOTE: all values are UPPERCASED!"""
return {
"",
" ",
" ",
"NONE",
"NA",
"NULL",
"NAN",
# CUSTOM
"#", # Parcel Match dataset has useless a_sunit=="#" values!
"%", # reserve/SPECIAL ready pakey delimiter
"HAS NONE",
"NOT AVAILABLE",
"N/A",
"-", # some SBO data just contains this value for empty string!
"<NULL>", # location_id col in ND SBO CAI vol0 file
}
[docs]
@staticmethod
def helper_gen_poi_id(indf, md5_cols, prepcols, res_idcol=None):
"""
prepcols is a subset of md5_cols (should only prep cols that goes into making of poi_id)
LATER, only need to prep place_name; as pakey had been prepped; and all other cols must be of enum type!
e.g. helper_gen_poi_id(finaldf, "created_by datasource pakey place_name ptype_mid".split(), "place_name".split())
"""
# md5_cols = "created_by datasource pakey place_name ptype_mid".split()
# for col in "created_by datasource place_name".split():
for col in prepcols:
indf[col] = (
indf[col].str.strip().str.upper().fillna("")
) # .str.replace("%", "")
from hashlib import md5
if res_idcol is None:
res_idcol = "poi_id"
indf[res_idcol] = (
indf[md5_cols]
.fillna("")
.apply(lambda row: "@".join(row.values.astype(str)), axis=1)
.apply(lambda x: md5(str(x).encode("utf-8")).hexdigest())
)
return indf
[docs]
@staticmethod
def helper_clean_addr_gen_pakey(
indf,
STREET_COL,
CITY_COL,
ZIPFULL_COL,
STATE_COL,
fill_statecode=None,
newpakeycol="clean_fakey",
):
"""GENERATE pakey as 'clean_fakey', and also validate address components
RETURN new columns: clean_addr_street clean_addr_city clean_addr_zip5 clean_addr_zipplus4 clean_addr_state
and newpakeycol/clean_fakey
"""
import numpy as np
STRING_NULLS = RADUtil.get_null_string_uppercased()
SF57R_ls = list(Config.STATE_FIPS.keys())
indf = indf.copy()
if fill_statecode is not None:
assert fill_statecode in set(SF57R_ls), f"Invalid {fill_statecode=}"
indf["clean_addr_state"] = fill_statecode.upper()
else:
assert STATE_COL in set(indf), f"MISSING COLUMN {STATE_COL=}"
indf["clean_addr_state"] = (
indf[STATE_COL]
.str.strip()
.str.upper()
.replace("%", "")
.replace(r"\s+", " ", regex=True)
.fillna("ZX")
)
assert all(
indf["clean_addr_state"].isin(SF57R_ls)
), "There are INVALID STATE_COL values!"
# assert all(indf.clean_addr_state.dropna().isin(enum_statecode57))
temp_series = (
indf[ZIPFULL_COL]
.fillna("")
.astype(str)
.str.upper()
.str.strip()
.replace(STRING_NULLS, "")
.copy()
)
## REMOVE and REPLACE invalid zipcode values
indf["clean_addr_zip5"] = temp_series.str[:5].copy()
indf["clean_addr_zip5"] = np.where(
(indf["clean_addr_zip5"].str.isdigit())
& (indf["clean_addr_zip5"].str.len() == 5),
indf["clean_addr_zip5"],
np.nan,
)
# DONT HAVE TO CHECK: assert all(indf.clean_addr_zip5.dropna().str.len() == 5)
# DONT HAVE TO CHECK: assert all(indf.clean_addr_zip5.dropna().str.isdigit())
indf["clean_addr_zipplus4"] = np.where(
temp_series.str.len().isin({9, 10}), temp_series.str[-4:], np.nan
)
## CHECK zipcodes are digits
assert all(indf.clean_addr_zipplus4.dropna().str.len() == 4)
assert all(indf.clean_addr_zipplus4.dropna().str.isdigit())
## CONVERT "" and other NULL forms to NaN
indf["clean_addr_street"] = indf[STREET_COL].copy()
indf["clean_addr_city"] = indf[CITY_COL].copy()
for (
col
) in "clean_addr_street clean_addr_city clean_addr_zip5 clean_addr_zipplus4".split():
indf[col] = (
indf[col]
.str.strip()
.str.upper()
.fillna("")
.str.replace("%", "")
.str.replace(r"\s+", " ", regex=True)
.replace(STRING_NULLS, np.nan)
)
## FILL NULL to make pakey
indf[newpakeycol] = (
indf.clean_addr_street.fillna("")
+ "%"
+ indf.clean_addr_city.fillna("")
+ "%"
+ indf.clean_addr_zip5.fillna("")
+ "%"
+ indf.clean_addr_state
)
return indf