Spaces:
Build error
Build error
File size: 2,773 Bytes
7b6ee0f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
import pandas as pd
import string
def get_function_mapping(df_metadata, split_ref):
df_cp = df_metadata[['function', 'desc', 'split']].copy()
df_cp.drop_duplicates(inplace=True)
return {function:desc for function, desc, split in df_cp.values if split==split_ref}
def get_channel_mapping(df_metadata):
df_cp = df_metadata[['channel', 'channel_desc']].copy()
df_cp.drop_duplicates(inplace=True)
return {channel:desc for channel, desc in df_cp.values}
def get_valid_field(df_metadata):
df_cp = df_metadata[["function", "field"]].copy()
df_cp.drop_duplicates(inplace=True)
field_list = df_cp.field.tolist()
function_list = df_cp.function.tolist()
field_list_out=[]
field_mapping={}
for temp_field, temp_function in zip(field_list, function_list):
temp_field = temp_field.split(" ### ")
temp_field = [x.strip().lower() for x in temp_field if x.strip() != '']
temp_field = [x.translate(str.maketrans(' ', '_', string.punctuation)) for x in temp_field]
for item in temp_field:
if item not in field_list_out:
field_list_out.append(item)
temp_field = "(" + ", ".join(temp_field) + ")"
field_mapping[temp_function] = temp_field
return field_list_out, field_mapping
def get_channel_to_function_mapping(df_metadata):
df_cp = df_metadata[["channel", "function"]].copy()
df_cp.drop_duplicates(inplace=True)
out_dict={}
for temp_channel, temp_function in df_cp.values:
if temp_channel not in out_dict.keys():
out_dict[temp_channel] = []
temp_function = temp_function.split(".")[-1]
out_dict[temp_channel].append(temp_function)
out_dict = {k:", ".join(v) for k,v in out_dict.items()}
return out_dict
def get_metadata(path):
df_metadata = pd.read_csv(path)
df_metadata = df_metadata.fillna('')
function_dict_trigger = get_function_mapping(df_metadata=df_metadata, split_ref='trigger')
function_dict_action = get_function_mapping(df_metadata=df_metadata, split_ref='action')
channel_dict = get_channel_mapping(df_metadata=df_metadata)
valid_field,field_mapping = get_valid_field(df_metadata=df_metadata)
channel_to_function_dict = get_channel_to_function_mapping(df_metadata=df_metadata)
return channel_dict, function_dict_trigger, function_dict_action, field_mapping, valid_field, channel_to_function_dict
def append_prefix(desc, prefix):
return prefix + desc
def append_suffix(desc, suffix):
return desc + suffix
def process_field(raw_field):
field = raw_field.split(" ### ")
field = [x.strip().lower() for x in field if x.strip() != '']
field = [x.translate(str.maketrans(' ', '_', string.punctuation)) for x in field]
return field |