Untitled
unknown
plain_text
2 months ago
2.6 kB
5
Indexable
import pandas as pd import sqlite3 import ast class ProgramProcessor: def __init__(self, db_path): self.conn = sqlite3.connect(db_path) def find_valid_rows(self, df, thresholds=[50, 45, 40, 35, 30, 25, 20, 15, 10, 5]): return next((df[df.apply(lambda row: (row['Sequence'] - row['Conditional_Seq']) > t and df[(df['Sequence'] >= row['Conditional_Seq']) & (df['Sequence'] <= row['Sequence'])]['Target'].notna().any(), axis=1)] for t in thresholds if not df.empty), pd.DataFrame()) @staticmethod def is_literal(value): try: return isinstance(ast.literal_eval(value), str) except (ValueError, SyntaxError): return False @staticmethod def filter_and_select(df): return df[~df['Target'].str.contains(r'\bOF\b', regex=True, na=False)] or df.loc[[df['Conditional_Seq'].idxmin()]] def process_program(self, program): data = pd.read_sql(f"SELECT * FROM BRE_LOGIC_TBL WHERE Program_Name='{program}'", self.conn) fil = self.find_valid_rows(data) if fil.empty: return None start, end = fil.loc[fil.eval('Sequence - Conditional_Seq').idxmax(), ['Conditional_Seq', 'Sequence']] fil_df = data.query("@start <= Sequence <= @end") fil_df[['Source', 'Target']] = fil_df[['Source', 'Target']].applymap(lambda x: ast.literal_eval(x) if isinstance(x, str) and x else None) seq_list = fil_df.loc[fil_df['Keywords'].isin(['IF', 'ELSE', 'EVALUATE', 'WHEN']), 'Sequence'] fil_df_3 = fil_df[fil_df[['Source', 'Target']].notna().all(axis=1) & fil_df['Conditional_Seq'].isin(seq_list)] for col in ['Source', 'Target']: if fil_df_3.empty: fil_df_3 = fil_df[~fil_df[col].apply(lambda x: self.is_literal(x[0]) if x else False)] if fil_df_3.empty: return None result = self.filter_and_select(fil_df_3) result['Program_Name'] = program return pd.DataFrame([result]) if isinstance(result, pd.Series) else result def process_multiple_programs(self, programs): return pd.concat(filter(None, map(self.process_program, programs)), ignore_index=True) or pd.DataFrame() # Usage db_path = "/c01/home/lidtd3o/APP_K/app_db_pc.db" programs = ["PAJSFN20", "OTHER_PROG1", "OTHER_PROG2"] processor = ProgramProcessor(db_path) final_result_df = processor.process_multiple_programs(programs) print(final_result_df)
Editor is loading...
Leave a Comment