| # import pandas as pd | |
| # import os | |
| # import re | |
| # from huggingface_hub import InferenceClient | |
| # class DataProcessor: | |
| # INTERVENTION_COLUMN = 'Did the intervention happen today?' | |
| # ENGAGED_STR = 'Engaged (Respect, Responsibility, Effort)' | |
| # PARTIALLY_ENGAGED_STR = 'Partially Engaged (about 50%)' | |
| # NOT_ENGAGED_STR = 'Not Engaged (less than 50%)' | |
| # def __init__(self): | |
| # self.hf_api_key = os.getenv('HF_API_KEY') | |
| # if not self.hf_api_key: | |
| # raise ValueError("HF_API_KEY not set in environment variables") | |
| # self.client = InferenceClient(api_key=self.hf_api_key) | |
| # def read_excel(self, uploaded_file): | |
| # return pd.read_excel(uploaded_file) | |
| # def format_session_data(self, df): | |
| # df['Date of Session'] = self.safe_convert_to_datetime(df['Date of Session'], '%m/%d/%Y') | |
| # df['Timestamp'] = self.safe_convert_to_datetime(df['Timestamp'], '%I:%M %p') | |
| # df['Session Start Time'] = self.safe_convert_to_time(df['Session Start Time'], '%I:%M %p') | |
| # df['Session End Time'] = self.safe_convert_to_time(df['Session End Time'], '%I:%M %p') | |
| # df = df[['Date of Session', 'Timestamp'] + [col for col in df.columns if col not in ['Date of Session', 'Timestamp']]] | |
| # return df | |
| # def safe_convert_to_time(self, series, format_str='%I:%M %p'): | |
| # try: | |
| # converted = pd.to_datetime(series, format='%H:%M:%S', errors='coerce') | |
| # if format_str: | |
| # return converted.dt.strftime(format_str) | |
| # return converted | |
| # except Exception as e: | |
| # print(f"Error converting series to time: {e}") | |
| # return series | |
| # def safe_convert_to_datetime(self, series, format_str=None): | |
| # try: | |
| # converted = pd.to_datetime(series, errors='coerce') | |
| # if format_str: | |
| # return converted.dt.strftime(format_str) | |
| # return converted | |
| # except Exception as e: | |
| # print(f"Error converting series to datetime: {e}") | |
| # return series | |
| # def replace_student_names_with_initials(self, df): | |
| # updated_columns = [] | |
| # for col in df.columns: | |
| # if col.startswith('Student Attendance'): | |
| # match = re.match(r'Student Attendance \[(.+?)\]', col) | |
| # if match: | |
| # name = match.group(1) | |
| # name_parts = name.split() | |
| # if len(name_parts) == 1: | |
| # initials = name_parts[0][0] | |
| # else: | |
| # initials = ''.join([part[0] for part in name_parts]) | |
| # updated_columns.append(f'Student Attendance [{initials}]') | |
| # else: | |
| # updated_columns.append(col) | |
| # else: | |
| # updated_columns.append(col) | |
| # df.columns = updated_columns | |
| # return df | |
| # def compute_intervention_statistics(self, df): | |
| # total_days = len(df) | |
| # sessions_held = df[self.INTERVENTION_COLUMN].str.strip().str.lower().eq('yes').sum() | |
| # sessions_not_held = df[self.INTERVENTION_COLUMN].str.strip().str.lower().eq('no').sum() | |
| # intervention_frequency = (sessions_held / total_days) * 100 if total_days > 0 else 0 | |
| # intervention_frequency = round(intervention_frequency, 0) | |
| # stats = { | |
| # 'Intervention Frequency (%)': [intervention_frequency], | |
| # 'Intervention Sessions Held': [sessions_held], | |
| # 'Intervention Sessions Not Held': [sessions_not_held], | |
| # 'Total Number of Days Available': [total_days] | |
| # } | |
| # return pd.DataFrame(stats) | |
| # def compute_student_metrics(self, df): | |
| # intervention_df = df[df[self.INTERVENTION_COLUMN].str.strip().str.lower() == 'yes'] | |
| # intervention_sessions_held = len(intervention_df) | |
| # student_columns = [col for col in df.columns if col.startswith('Student Attendance')] | |
| # student_metrics = {} | |
| # for col in student_columns: | |
| # student_name = col.replace('Student Attendance [', '').replace(']', '').strip() | |
| # student_data = intervention_df[[col]].copy() | |
| # student_data[col] = student_data[col].fillna('Absent') | |
| # attendance_values = student_data[col].apply(lambda x: 1 if x in [ | |
| # self.ENGAGED_STR, | |
| # self.PARTIALLY_ENGAGED_STR, | |
| # self.NOT_ENGAGED_STR | |
| # ] else 0) | |
| # sessions_attended = attendance_values.sum() | |
| # attendance_pct = (sessions_attended / intervention_sessions_held) * 100 if intervention_sessions_held > 0 else 0 | |
| # attendance_pct = round(attendance_pct) | |
| # engagement_counts = { | |
| # 'Engaged': 0, | |
| # 'Partially Engaged': 0, | |
| # 'Not Engaged': 0, | |
| # 'Absent': 0 | |
| # } | |
| # for x in student_data[col]: | |
| # if x == self.ENGAGED_STR: | |
| # engagement_counts['Engaged'] += 1 | |
| # elif x == self.PARTIALLY_ENGAGED_STR: | |
| # engagement_counts['Partially Engaged'] += 1 | |
| # elif x == self.NOT_ENGAGED_STR: | |
| # engagement_counts['Not Engaged'] += 1 | |
| # else: | |
| # engagement_counts['Absent'] += 1 # Count as Absent if not engaged | |
| # # Calculate percentages for engagement states | |
| # total_sessions = sum(engagement_counts.values()) | |
| # # Engagement (%) | |
| # engagement_pct = (engagement_counts['Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| # engagement_pct = round(engagement_pct) | |
| # engaged_pct = (engagement_counts['Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| # engaged_pct = round(engaged_pct) | |
| # partially_engaged_pct = (engagement_counts['Partially Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| # partially_engaged_pct = round(partially_engaged_pct) | |
| # not_engaged_pct = (engagement_counts['Not Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| # not_engaged_pct = round(not_engaged_pct) | |
| # absent_pct = (engagement_counts['Absent'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| # absent_pct = round(absent_pct) | |
| # # Store metrics in the required order | |
| # student_metrics[student_name] = { | |
| # 'Attendance (%)': attendance_pct, | |
| # 'Attendance #': sessions_attended, # Raw number of sessions attended | |
| # 'Engagement (%)': engagement_pct, | |
| # 'Engaged (%)': engaged_pct, | |
| # 'Partially Engaged (%)': partially_engaged_pct, | |
| # 'Not Engaged (%)': not_engaged_pct, | |
| # 'Absent (%)': absent_pct | |
| # } | |
| # # Create a DataFrame from student_metrics | |
| # student_metrics_df = pd.DataFrame.from_dict(student_metrics, orient='index').reset_index() | |
| # student_metrics_df.rename(columns={'index': 'Student'}, inplace=True) | |
| # return student_metrics_df | |
| # def compute_average_metrics(self, student_metrics_df): | |
| # # Calculate the attendance and engagement average percentages across students | |
| # attendance_avg_stats = student_metrics_df['Attendance (%)'].mean() # Calculate the average attendance percentage | |
| # engagement_avg_stats = student_metrics_df['Engagement (%)'].mean() # Calculate the average engagement percentage | |
| # # Round the averages to make them whole numbers | |
| # attendance_avg_stats = round(attendance_avg_stats) | |
| # engagement_avg_stats = round(engagement_avg_stats) | |
| # return attendance_avg_stats, engagement_avg_stats | |
| import pandas as pd | |
| import os | |
| import re | |
| from huggingface_hub import InferenceClient | |
| from graphviz import Digraph | |
| class DataProcessor: | |
| INTERVENTION_COLUMN = 'Did the intervention happen today?' | |
| ENGAGED_STR = 'Engaged (Respect, Responsibility, Effort)' | |
| PARTIALLY_ENGAGED_STR = 'Partially Engaged (about 50%)' | |
| NOT_ENGAGED_STR = 'Not Engaged (less than 50%)' | |
| def __init__(self, student_metrics_df=None): | |
| self.hf_api_key = os.getenv('HF_API_KEY') | |
| if not self.hf_api_key: | |
| raise ValueError("HF_API_KEY not set in environment variables") | |
| self.client = InferenceClient(api_key=self.hf_api_key) | |
| self.student_metrics_df = student_metrics_df | |
| def read_excel(self, uploaded_file): | |
| return pd.read_excel(uploaded_file) | |
| def format_session_data(self, df): | |
| df['Date of Session'] = self.safe_convert_to_datetime(df['Date of Session'], '%m/%d/%Y') | |
| df['Timestamp'] = self.safe_convert_to_datetime(df['Timestamp'], '%I:%M %p') | |
| df['Session Start Time'] = self.safe_convert_to_time(df['Session Start Time'], '%I:%M %p') | |
| df['Session End Time'] = self.safe_convert_to_time(df['Session End Time'], '%I:%M %p') | |
| return df | |
| def safe_convert_to_time(self, series, format_str='%I:%M %p'): | |
| try: | |
| converted = pd.to_datetime(series, format='%H:%M:%S', errors='coerce') | |
| if format_str: | |
| return converted.dt.strftime(format_str) | |
| return converted | |
| except Exception as e: | |
| print(f"Error converting series to time: {e}") | |
| return series | |
| def safe_convert_to_datetime(self, series, format_str=None): | |
| try: | |
| converted = pd.to_datetime(series, errors='coerce') | |
| if format_str: | |
| return converted.dt.strftime(format_str) | |
| return converted | |
| except Exception as e: | |
| print(f"Error converting series to datetime: {e}") | |
| return series | |
| def replace_student_names_with_initials(self, df): | |
| updated_columns = [] | |
| for col in df.columns: | |
| if col.startswith('Student Attendance'): | |
| match = re.match(r'Student Attendance \[(.+?)\]', col) | |
| if match: | |
| name = match.group(1) | |
| initials = ''.join([part[0] for part in name.split()]) | |
| updated_columns.append(f'Student Attendance [{initials}]') | |
| else: | |
| updated_columns.append(col) | |
| else: | |
| updated_columns.append(col) | |
| df.columns = updated_columns | |
| return df | |
| def compute_intervention_statistics(self, df): | |
| total_days = len(df) | |
| sessions_held = df[self.INTERVENTION_COLUMN].str.strip().str.lower().eq('yes').sum() | |
| intervention_frequency = (sessions_held / total_days) * 100 if total_days > 0 else 0 | |
| return pd.DataFrame({ | |
| 'Intervention Frequency (%)': [round(intervention_frequency, 0)], | |
| 'Intervention Sessions Held': [sessions_held], | |
| 'Intervention Sessions Not Held': [total_days - sessions_held], | |
| 'Total Number of Days Available': [total_days] | |
| }) | |
| def compute_student_metrics(self, df): | |
| intervention_df = df[df[self.INTERVENTION_COLUMN].str.strip().str.lower() == 'yes'] | |
| intervention_sessions_held = len(intervention_df) | |
| student_columns = [col for col in df.columns if col.startswith('Student Attendance')] | |
| student_metrics = {} | |
| for col in student_columns: | |
| student_name = col.replace('Student Attendance [', '').replace(']', '').strip() | |
| student_data = intervention_df[[col]].copy() | |
| student_data[col] = student_data[col].fillna('Absent') | |
| attendance_values = student_data[col].apply(lambda x: 1 if x in [ | |
| self.ENGAGED_STR, | |
| self.PARTIALLY_ENGAGED_STR, | |
| self.NOT_ENGAGED_STR | |
| ] else 0) | |
| sessions_attended = attendance_values.sum() | |
| attendance_pct = (sessions_attended / intervention_sessions_held) * 100 if intervention_sessions_held > 0 else 0 | |
| attendance_pct = round(attendance_pct) | |
| engagement_counts = { | |
| 'Engaged': 0, | |
| 'Partially Engaged': 0, | |
| 'Not Engaged': 0, | |
| 'Absent': 0 | |
| } | |
| for x in student_data[col]: | |
| if x == self.ENGAGED_STR: | |
| engagement_counts['Engaged'] += 1 | |
| elif x == self.PARTIALLY_ENGAGED_STR: | |
| engagement_counts['Partially Engaged'] += 1 | |
| elif x == self.NOT_ENGAGED_STR: | |
| engagement_counts['Not Engaged'] += 1 | |
| else: | |
| engagement_counts['Absent'] += 1 # Count as Absent if not engaged | |
| # Calculate percentages for engagement states | |
| total_sessions = sum(engagement_counts.values()) | |
| # Engagement (%) | |
| engagement_pct = (engagement_counts['Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| engagement_pct = round(engagement_pct) | |
| engaged_pct = (engagement_counts['Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| engaged_pct = round(engaged_pct) | |
| partially_engaged_pct = (engagement_counts['Partially Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| partially_engaged_pct = round(partially_engaged_pct) | |
| not_engaged_pct = (engagement_counts['Not Engaged'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| not_engaged_pct = round(not_engaged_pct) | |
| absent_pct = (engagement_counts['Absent'] / total_sessions * 100) if total_sessions > 0 else 0 | |
| absent_pct = round(absent_pct) | |
| # Determine if the student attended ≥ 90% of sessions | |
| attended_90 = "Yes" if attendance_pct >= 90 else "No" | |
| # Determine if the student was engaged ≥ 80% of the time | |
| engaged_80 = "Yes" if engaged_pct >= 80 else "No" | |
| # Store metrics in the required order | |
| student_metrics[student_name] = { | |
| 'Attended ≥ 90%': attended_90, | |
| 'Engagement ≥ 80%': engaged_80, | |
| 'Attendance (%)': attendance_pct, | |
| # 'Attendance #': sessions_attended, | |
| 'Engagement (%)': engagement_pct, | |
| 'Engaged (%)': engaged_pct, | |
| 'Partially Engaged (%)': partially_engaged_pct, | |
| 'Not Engaged (%)': not_engaged_pct, | |
| 'Absent (%)': absent_pct | |
| } | |
| # Create a DataFrame from student_metrics | |
| student_metrics_df = pd.DataFrame.from_dict(student_metrics, orient='index').reset_index() | |
| student_metrics_df.rename(columns={'index': 'Student'}, inplace=True) | |
| return student_metrics_df | |
| def compute_average_metrics(self, student_metrics_df): | |
| # Calculate the attendance and engagement average percentages across students | |
| attendance_avg_stats = student_metrics_df['Attendance (%)'].mean() # Calculate the average attendance percentage | |
| engagement_avg_stats = student_metrics_df['Engagement (%)'].mean() # Calculate the average engagement percentage | |
| # Round the averages to make them whole numbers | |
| attendance_avg_stats = round(attendance_avg_stats) | |
| engagement_avg_stats = round(engagement_avg_stats) | |
| return attendance_avg_stats, engagement_avg_stats | |
| def evaluate_student(self, row, attendance_threshold=90, engagement_threshold=80): | |
| if row["Attended ≥ 90%"] == "No": | |
| return "Address Attendance" | |
| elif row["Engagement ≥ 80%"] == "No": | |
| return "Address Engagement" | |
| return "Consider addressing logistical barriers, improving fidelity, and/or collecting progress monitoring data" | |
| # def build_tree_diagram(self, row): | |
| # dot = Digraph() | |
| # dot.node("Q1", "Has the student attended ≥ 90% of interventions?") | |
| # dot.node("Q2", "Has the student been engaged ≥ 80% of intervention time?") | |
| # dot.node("A1", "Address Attendance", shape="box") | |
| # dot.node("A2", "Address Engagement", shape="box") | |
| # dot.node("A3", "Consider addressing logistical barriers", shape="box") | |
| # if row["Attended ≥ 90%"] == "No": | |
| # dot.edge("Q1", "A1", label="No") | |
| # else: | |
| # dot.edge("Q1", "Q2", label="Yes") | |
| # dot.edge("Q2", "A2" if row["Engagement ≥ 80%"] == "No" else "A3", label="Yes") | |
| # return dot | |