| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- import pandas as pd
- from pandas import DataFrame
- import numpy as np
- # import dtale
- def get_agg_pdu_info_by_controller(pdu_name: str, pdu_triggering_df: DataFrame):
- # select pdu_name releated pdus in Pdu Triggering
- pdu_triggering_df.fillna('', inplace=True)
- criterion_select_pdus = pdu_triggering_df['Pdu Name'].str.contains(pdu_name)
- releated_pdus = pdu_triggering_df[criterion_select_pdus]
- pdu_info_agg = releated_pdus \
- .groupby('Controller') \
- .agg(
- PduName = ('Pdu Name', lambda x: ','.join(x.unique())),
- PortDirection = ('Port Direction', lambda x: ','.join(x.unique())),
- signal=('Signal Trigger', lambda x: ','.join(x))
- ) \
- .reset_index()
- return pdu_info_agg
- def get_normal_or_contained_pdu(pdu_name: str, aggregated_pdu_info: DataFrame, container_df: DataFrame):
- unique_pdu_list = aggregated_pdu_info['PduName'].unique()
- # get contained pdu's name, and add pdu name as normal can pdu name.
- cri_is_contained = container_df['Contained PDU'].apply(lambda x: x[5:]).isin(unique_pdu_list) # trip head 'PduTr'
- normal_contained_pdus = np.append(container_df[cri_is_contained]['PDU Name'].unique(), pdu_name)
- return normal_contained_pdus
- def get_frame_info_by_pdu_name_list(normal_contained_pdus, frame_df: DataFrame):
- cri_has_frame_info = frame_df['Pdu Trigger'].apply(lambda x: x[5:]).isin(normal_contained_pdus) # trip head 'PduTr'
- frame_info = frame_df[cri_has_frame_info]
- return frame_info
- def get_useful_info_by_pdu_name(pdu_name: str, frame_df, pdu_triggering_df, container_df, df_idpu):
- pdu_info_agg = get_agg_pdu_info_by_controller(pdu_name, pdu_triggering_df)
- # print(pdu_info_agg)
- normal_contained_pdus = get_normal_or_contained_pdu(pdu_name, pdu_info_agg, container_df)
- # print(normal_contained_pdus)
- frame_info = get_frame_info_by_pdu_name_list(normal_contained_pdus, frame_df)
- # print(frame_info.loc[:, ['Controller', 'Frame Name', 'Frame ID', 'Pdu Trigger']])
- # pdu_info_merged = pd.merge(pdu_info_agg, frame_info.loc[:, ['Frame Name', 'Frame ID', 'Pdu Trigger']], on='Controller')
-
- contained_info = get_contained_pdu_info(pdu_name, df_idpu)
-
- pdu_info_merged = pd.merge(pdu_info_agg, frame_info.loc[:, ['Controller', 'Frame Name', 'Frame ID', 'Pdu Trigger']], on='Controller')
- pdu_info_merged = pd.merge(pdu_info_merged, contained_info, on='PduName', how='outer')
- return pdu_info_merged
- def get_contained_pdu_info(pdu_name: str, ipdu_df):
- df_ipdu_contained = ipdu_df[ipdu_df['Contained PDU ID'].notna()]
- df_contained_info = df_ipdu_contained \
- .groupby('PDU Name') \
- .agg(
- ContainedPduID = ('Contained PDU ID', lambda x: ','.join(x.unique())),
- ) \
- .reset_index()
- df_contained_info.rename(columns={'PDU Name': 'PduName'}, inplace=True)
- # print(df_contained_info)
- criterion_select_pdus = df_contained_info['PduName'].str.contains(pdu_name)
- relate_contained_pdu = df_contained_info[criterion_select_pdus]
-
- return relate_contained_pdu
- def get_pdu_info_by_signal_name(signal_name: str, pdu_triggering_df):
- criterion_select_pdus = pdu_triggering_df['Signal Trigger'].str.contains(signal_name)
- releated_pdus = pdu_triggering_df[criterion_select_pdus]
-
- return releated_pdus
- def test():
- xlsx = pd.ExcelFile('OutputFile_Attributes_MA_20220921-qy1_EP39 EREV_SIMU+_GW-arxml-V01.xlsx')
- # print(xlsx.sheet_names)
- df_idpu = pd.read_excel(xlsx, sheet_name='IPdu', index_col=0)
- df_frame = pd.read_excel(xlsx, sheet_name='FrameTriggering', index_col=0)
- df_pdu_trig = pd.read_excel(xlsx, sheet_name='PduTriggering', index_col=0)
- df_container = pd.read_excel(xlsx, sheet_name='ContainerPdu', index_col=0)
- ipdu_dtyps = {'Contained PDU ID':str}
- df_ipdu = pd.read_excel(xlsx, sheet_name='IPdu', index_col=0, dtype=ipdu_dtyps)
-
- pdu_name = 'CCP_020ms_PDU02'
- #pdu_name = 'ECM_010ms_PDU00'
- # pdu_name = 'FVCM_ADCANFD_Sporadic_Container12'
- # pdu_name = 'SCS_020ms_PDU04'
- search_info = get_useful_info_by_pdu_name(pdu_name, df_frame, df_pdu_trig, df_container, df_ipdu)
- # dtale.show(search_info,open_browser=True)#
- print(search_info)
-
- signal_name = 'FICMACD'
- search_pdu_info = get_pdu_info_by_signal_name(signal_name, df_pdu_trig)
- print(search_pdu_info)
|
