The Personnel Information (PersonnelInformation) Data Block in Assette retrieves active personnel details as of a specified date. It allows users to apply filters such as gender, job function, management status, investment professional status, ethnicity, and founder status. This block consolidates personnel metadata and experience calculations, returning enriched personnel profiles with structured details on job roles, tenure, experience, and organizational attributes.
PeopleMetadata = read("qualitativeMetadata",{"type": "People"})["data"]
field_title_dict = {}
for pdict in PeopleMetadata["fields"]:
field_title_dict[pdict["name"]] = pdict["title"]
fields = list(field_title_dict.keys())
if "fields" in params and params["fields"]:
# templist = params["fields"].split('|')
templist = params["fields"].replace('|', ',').split(',')
for item in templist:
if item not in fields:
fields.append(item)
fields = "|".join(fields)
data = read("GetPersonnelInfo",{"fields": fields,"getdataasofnow": params["getdataasofnow"],"asofdate": params["AsofDate"]})["data"]
df = pd.DataFrame(data)
as_of_date = pd.to_datetime(params["AsofDate"], errors="coerce")
getdataasofnow = params["getdataasofnow"]
if (isinstance(getdataasofnow, str) and getdataasofnow.lower() == "yes") or getdataasofnow is True:
current_date = datetime.datetime.now()
as_of_date = pd.to_datetime(current_date, errors="coerce")
def convert_days_to_ymd(days):
start_date = datetime(1, 1, 1)
end_date = start_date + timedelta(days=int(days))
years = end_date.year - start_date.year
months = end_date.month - start_date.month
days = end_date.day - start_date.day
if days < 0:
months -= 1
days += (start_date + timedelta(days=30)).day
if months < 0:
years -= 1
months += 12
return f"{years} Year(s) {months} Month(s) and {days} Day(s)"
# Check if the DataFrame is not empty before applying conditions
if not df.empty:
# Filter the DataFrame based on the conditions
df = df[
((df['firmStartDate'].isna() | (pd.to_datetime(df['firmStartDate'], errors='coerce', format='%m/%d/%Y') < as_of_date)) &
((df['terminationDate'].isna()) | (pd.to_datetime(df['terminationDate'], errors='coerce', format='%m/%d/%Y') > as_of_date) | ('isEmployeeStillAtCompany' in df.columns and df['isEmployeeStillAtCompany'] == 'Yes')))
]
columns_to_convert = [
"totalExperience",
"totalIndustryExperience",
"totalInvestmentExperience",
"totalProfessionalExperience",
"totalExperienceAtCurrentOrganization",
"continuousExperienceAtCurrentOrganization",
"previousStintsAtCurrentOrganization"
]
for column in columns_to_convert:
if column in df.columns:
# df[column] = df[column].apply(lambda days: (lambda start_date, end_date: f"{end_date.year - start_date.year - ((end_date.month, end_date.day) < (start_date.month, start_date.day))} Year(s) {((end_date.month - start_date.month - (end_date.day < start_date.day)) % 12)} Month(s) and {(end_date - (start_date + timedelta(days=(end_date - start_date).days // 365 * 365 + (end_date - start_date).days % 365 // 30 * 30))).days} Day(s)")(
# datetime(1, 1, 1), datetime(1, 1, 1) + datetime.timedelta(days=int(days))
# ))
# df[column] = df[column].apply(convert_days_to_ymd)
# df[column] = df[column].apply(convert_days_to_years_months_days)
df[column] = df[column].apply(lambda days: f"{int(days) // 365} Year(s) {(int(days) % 365) // 30} Month(s) and {(int(days) % 365) % 30} Day(s)")
else:
# If DataFrame is empty, create an empty DataFrame with the same columns
df = pd.DataFrame(columns=df.columns)
def filter_row(row, params):
# Check if the row matches all specified criteria
return (
(params["gender"] == "All" or ('gender' in row and row["gender"] == params["gender"]))
and (params["investmentProfessional"] == "All" or ('investmentProfessional' in row and row["investmentProfessional"] == params["investmentProfessional"]))
and (params["management"] == "All" or ('management' in row and row["management"] == params["management"]))
and (params["ethnicity"] == "All" or ('ethnicity' in row and str(row["ethnicity"]) in params["ethnicity"] and str(row["ethnicity"]) != ""))
and (params["isTheEmployeeAFounderOfTheOrganization"] == "All" or ('isTheEmployeeAFounderOfTheOrganization' in row and row["isTheEmployeeAFounderOfTheOrganization"] == params["isTheEmployeeAFounderOfTheOrganization"]))
and (params["primaryJobFunction"] == "All" or
('primaryJobFunction' in row and str(row["primaryJobFunction"]) in params["primaryJobFunction"] and str(row["primaryJobFunction"]) != ""))
)
# Apply the filter function to the DataFrame and store the result in a new DataFrame
filtered_df = df[df.apply(filter_row, axis=1, params=params)]
# Reset the index of the filtered DataFrame
filtered_df.reset_index(drop=True, inplace=True)
final_result_list = []
for i in range(filtered_df.shape[0]):
row_dict = filtered_df.loc[i].to_dict()
flat_row = {}
for k, v in row_dict.items():
# Flatten dict
if isinstance(v, dict):
flat_row[k] = ', '.join([str(key) + ': ' + str(value) for key, value in v.items()])
# Flatten list of dicts
elif isinstance(v, list) and all(isinstance(item, dict) for item in v):
flat_row[k] = ' | '.join([', '.join([str(key) + ': ' + str(value) for key, value in item.items()]) for item in v])
# Flatten list of other types
elif isinstance(v, list):
flat_row[k] = ', '.join([str(item) for item in v])
else:
flat_row[k] = "" if v is None else str(v)
final_result_list.append(flat_row)
response["data"]=final_result_list
response["values"]=field_title_dict
