METECH-DEV/Plugit-Charging-Transactions-Convertor-v2
0
1import pandas as pd2import gradio as gr3from datetime import datetime, timedelta4import os5import tempfile6 7 8def process_charging_sessions(file_a, output_file, start_date, end_date):9 # Read CSV file10 df_a = pd.read_csv(file_a, sep=';', parse_dates=['timestampStart', 'timestampStop'])11 # Convert UTC timestamps to SGT (UTC+8)12 df_a['timestampStart'] = df_a['timestampStart'].dt.tz_convert('Asia/Singapore')13 df_a['timestampStop'] = df_a['timestampStop'].dt.tz_convert('Asia/Singapore')14 15 # Parse start and end dates16 start_date_obj = pd.to_datetime(start_date).date()17 end_date_obj = pd.to_datetime(end_date).date()18 19 # Filter data based on date range20 # Keep sessions that overlap with the requested date range21 df_a = df_a[22 (df_a['timestampStart'].dt.date <= end_date_obj) & 23 (df_a['timestampStop'].dt.date >= start_date_obj)24 ]25 26 # Get unique charge boxes and create a mapping of chargeBoxIdentity to power and chargePoint27 charge_boxes = df_a['chargeBoxIdentity'].unique()28 power_mapping = df_a.groupby('chargeBoxIdentity')['power'].first().to_dict()29 chargepoint_mapping = df_a.groupby('chargeBoxIdentity')['chargePoint'].first().to_dict()30 31 # Initialize output dataframe32 output_data = []33 34 # First, process actual charging sessions35 for _, session in df_a.iterrows():36 # Extract session information37 start_time = session['timestampStart']38 end_time = session['timestampStop']39 charge_box = session['chargeBoxIdentity']40 power = session['power']41 charge_point = session['chargePoint']42 total_energy = session['energy'] # Total energy in kWh43 duration_seconds = session['duration'] # Duration in seconds44 45 # Calculate energy per second46 if duration_seconds > 0:47 energy_per_second = total_energy / duration_seconds48 else:49 energy_per_second = 050 51 # Round start time down to the beginning of the hour52 current_block_start = start_time.replace(minute=0, second=0, microsecond=0)53 # Calculate end time of the last block54 end_block = end_time.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)55 56 # Process each hour in the session57 while current_block_start < end_block:58 # Skip blocks outside the requested date range59 if current_block_start.date() < start_date_obj or current_block_start.date() > end_date_obj:60 current_block_start += timedelta(hours=1)61 continue62 63 # Calculate time range for current block (XX:00:00 to XX:59:59.999...)64 block_start = current_block_start65 block_end = current_block_start + timedelta(hours=1)66 67 # Calculate actual charging time within this block68 charge_start = max(block_start, start_time)69 charge_end = min(block_end, end_time)70 71 # Calculate duration in seconds72 duration_seconds_in_block = (charge_end - charge_start).total_seconds()73 74 # Only add if there was actual charging in this block75 if duration_seconds_in_block > 0:76 # Calculate energy transferred in this hour77 energy_transferred = (energy_per_second * duration_seconds_in_block)/100078 79 # Calculate time utilization as a decimal (0 to 1) with 5 decimal places80 utilization = round(duration_seconds_in_block / 3600, 5) # Convert to fraction of hour81 82 # Format times as HH:MM83 start_time_str = block_start.strftime('%H:%M')84 end_time_str = (block_end - timedelta(seconds=1)).strftime('%H:%M')85 86 # Add to output87 output_data.append({88 'Date': block_start.strftime('%d %B %Y'),89 'Start Time (hh:mm)': start_time_str,90 'End Time (hh:mm)': end_time_str,91 'EVSE Connector ID number': charge_box,92 'EV Charge Point Power Rating (kW)': round(power / 1000, 1),93 'Energy transferred (kWh)': round(energy_transferred, 4),94 'Time utilisation (%)': utilization,95 'chargePoint': charge_point96 })97 98 current_block_start += timedelta(hours=1)99 100 # Create a set to track existing time blocks101 existing_blocks = set()102 for entry in output_data:103 block_key = (entry['Date'], entry['Start Time (hh:mm)'], entry['EVSE Connector ID number'])104 existing_blocks.add(block_key)105 106 # Add filler blocks for each day and each charge box107 current_date = start_date_obj108 while current_date <= end_date_obj:109 for hour in range(24):110 for charge_box in charge_boxes:111 # Create time strings112 start_time = f"{hour:02d}:00"113 end_time = f"{hour:02d}:59"114 date_str = current_date.strftime('%d %B %Y')115 116 # Check if this block already exists117 block_key = (date_str, start_time, charge_box)118 if block_key not in existing_blocks:119 # Add filler block with correct power from mapping120 output_data.append({121 'Date': date_str,122 'Start Time (hh:mm)': start_time,123 'End Time (hh:mm)': end_time,124 'EVSE Connector ID number': charge_box,125 'EV Charge Point Power Rating (kW)': round(power_mapping[charge_box] / 1000, 1),126 'Energy transferred (kWh)': 0.0000,127 'Time utilisation (%)': 0.000,128 'chargePoint': chargepoint_mapping[charge_box]129 })130 current_date += timedelta(days=1)131 132 # Create output dataframe and sort by date, time, and charge box133 df_output = pd.DataFrame(output_data)134 df_output['datetime'] = pd.to_datetime(df_output['Date'] + ' ' + df_output['Start Time (hh:mm)'], 135 format='%d %B %Y %H:%M')136 df_output = df_output.sort_values(['EVSE Connector ID number','datetime']).drop('datetime', axis=1)137 138 # Reorder columns to match desired output139 df_output = df_output[[140 'Date',141 'Start Time (hh:mm)',142 'End Time (hh:mm)',143 'EVSE Connector ID number',144 'EV Charge Point Power Rating (kW)',145 'Energy transferred (kWh)',146 'Time utilisation (%)',147 'chargePoint'148 ]]149 150 # Save to Excel with custom formatting151 with pd.ExcelWriter(output_file, engine='xlsxwriter') as writer:152 workbook = writer.book153 154 # Create "EV Charge Points" consolidated sheet155 # Group by Date, Start Time, End Time, EVSE Connector ID, and Power Rating156 # Calculate average energy transferred for each group157 ev_charge_points_data = df_output.groupby([158 'Date',159 'Start Time (hh:mm)',160 'End Time (hh:mm)',161 'EVSE Connector ID number',162 'EV Charge Point Power Rating (kW)'163 ]).agg({164 'Energy transferred (kWh)': lambda x: round(x.mean(), 4), # Average energy to 4dp165 'Time utilisation (%)': lambda x: round(x.mean(), 5) # Average utilization to 5dp (displays as 3dp when formatted as %)166 }).reset_index()167 168 # Rename columns to match required format169 ev_charge_points_data = ev_charge_points_data.rename(columns={170 'EVSE Connector ID number': 'EVSE Registration Code (ERC) number',171 'Energy transferred (kWh)': 'Average energy transferred (kWh)',172 'Time utilisation (%)': 'Time utilisation (%) (Non-publicly accessible EVCP only)'173 })174 175 # Sort by ERC number as requested (and then by date/time for consistency)176 ev_charge_points_data = ev_charge_points_data.sort_values([177 'EVSE Registration Code (ERC) number',178 'Date', 179 'Start Time (hh:mm)'180 ])181 182 # Reorder columns183 ev_charge_points_data = ev_charge_points_data[[184 'Date',185 'Start Time (hh:mm)',186 'End Time (hh:mm)',187 'EVSE Registration Code (ERC) number',188 'EV Charge Point Power Rating (kW)',189 'Average energy transferred (kWh)',190 'Time utilisation (%) (Non-publicly accessible EVCP only)'191 ]]192 193 # Write to Excel with formatting194 ev_charge_points_data.to_excel(writer, index=False, sheet_name='EV Charge Points')195 worksheet = writer.sheets['EV Charge Points']196 197 # Create formats with borders198 header_format = workbook.add_format({199 'bold': True,200 'align': 'center',201 'valign': 'vcenter',202 'border': 1203 })204 border_format = workbook.add_format({'border': 1})205 energy_format = workbook.add_format({'num_format': '0.0000', 'border': 1})206 percent_format = workbook.add_format({'num_format': '0.000%', 'border': 1})207 208 # Apply borders to all cells209 num_rows = len(ev_charge_points_data)210 num_cols = len(ev_charge_points_data.columns)211 212 # Write headers with borders213 for col in range(num_cols):214 worksheet.write(0, col, ev_charge_points_data.columns[col], header_format)215 216 # Write data cells with borders217 for row in range(num_rows):218 for col in range(num_cols):219 cell_value = ev_charge_points_data.iloc[row, col]220 221 # Apply appropriate format based on column222 if col == 5: # Average energy transferred column223 cell_format = energy_format224 elif col == 6: # Time utilisation column225 cell_format = percent_format226 else:227 cell_format = border_format228 229 worksheet.write(row + 1, col, cell_value, cell_format)230 231 # Now create separate worksheets for each chargePoint232 charge_points = df_output['chargePoint'].unique() # type: ignore[call-arg]233 234 for cp in charge_points:235 # Filter data for this charge point236 cp_data = df_output[df_output['chargePoint'] == cp].copy()237 238 # Get unique power ratings for this charge point239 power_ratings = sorted(cp_data['EV Charge Point Power Rating (kW)'].unique()) # type: ignore[call-arg]240 241 # Group by Date and Start Time to aggregate242 aggregated_data = []243 244 # Get all unique date-time combinations245 date_times = cp_data[['Date', 'Start Time (hh:mm)', 'End Time (hh:mm)']].drop_duplicates() # type: ignore[call-arg]246 247 for _, dt_row in date_times.iterrows():248 date = dt_row['Date']249 start_time = dt_row['Start Time (hh:mm)']250 end_time = dt_row['End Time (hh:mm)']251 252 # Filter data for this date-time block253 block_data = cp_data[254 (cp_data['Date'] == date) & 255 (cp_data['Start Time (hh:mm)'] == start_time)256 ]257 258 # Calculate total energy transferred259 # Round to 4 decimal places to match Excel display format260 total_energy = round(block_data['Energy transferred (kWh)'].sum(), 4)261 262 # Initialize row data263 row_data = {264 'Date': date,265 'Start Time (hh:mm)': start_time,266 'End Time (hh:mm)': end_time,267 'Carpark Code': '',268 'Postal Code': '',269 'Total energy transferred by all EV charge points in the carpark (kWh)': total_energy270 }271 272 # Calculate averages for each power rating273 for power in power_ratings:274 power_data = block_data[block_data['EV Charge Point Power Rating (kW)'] == power]275 276 if len(power_data) > 0:277 # Round average energy to 4 decimal places278 avg_energy = round(power_data['Energy transferred (kWh)'].mean(), 4)279 # Round average utilization to 3 decimal places (stored as decimal, not percentage)280 avg_utilization = round(power_data['Time utilisation (%)'].mean(), 5)281 else:282 avg_energy = 0.0283 avg_utilization = 0.0284 285 row_data[f'{power}kW_energy'] = avg_energy286 row_data[f'{power}kW_utilization'] = avg_utilization287 288 aggregated_data.append(row_data)289 290 # Create DataFrame for this charge point291 df_cp = pd.DataFrame(aggregated_data)292 293 # Sort by date and time294 df_cp['datetime'] = pd.to_datetime(df_cp['Date'] + ' ' + df_cp['Start Time (hh:mm)'], 295 format='%d %B %Y %H:%M')296 df_cp = df_cp.sort_values('datetime').drop('datetime', axis=1)297 298 # Reorder columns - base columns first, then dynamic columns299 base_columns = [300 'Date',301 'Start Time (hh:mm)',302 'End Time (hh:mm)',303 'Carpark Code',304 'Postal Code',305 'Total energy transferred by all EV charge points in the carpark (kWh)'306 ]307 308 # Add dynamic columns in order: energy columns first, then utilization columns309 dynamic_columns = []310 for power in power_ratings:311 dynamic_columns.append(f'{power}kW_energy')312 for power in power_ratings:313 dynamic_columns.append(f'{power}kW_utilization')314 315 df_cp = df_cp[base_columns + dynamic_columns]316 317 318 # Rename dynamic columns to have proper headers319 # (Logic moved to Excel writing section)320 321 # Write to Excel322 # Create a safe sheet name (Excel has 31 char limit and some char restrictions)323 sheet_name = str(cp)[:31].replace('/', '_').replace('\\', '_').replace('*', '_').replace('[', '_').replace(']', '_').replace(':', '_').replace('?', '_')324 325 # Write without headers first (we'll add custom headers)326 df_cp.to_excel(writer, index=False, sheet_name=sheet_name, startrow=2, header=False)327 worksheet = writer.sheets[sheet_name]328 329 # Create formats330 header_format = workbook.add_format({331 'bold': True,332 'align': 'center',333 'valign': 'vcenter',334 'text_wrap': True,335 'border': 1336 })337 338 border_format = workbook.add_format({'border': 1})339 energy_format_border = workbook.add_format({'num_format': '0.0000', 'border': 1})340 percent_format_border = workbook.add_format({'num_format': '0.000%', 'border': 1})341 342 num_rows = len(df_cp)343 num_cols = len(base_columns) + len(dynamic_columns)344 345 # Write two-row headers with merges346 # Row 0: Main headers (with merges for base columns and category headers)347 # Row 1: Sub-headers (power ratings)348 349 # Base columns (A-F): Merge rows 0 and 1350 base_headers = [351 'Date',352 'Start Time (hh:mm)',353 'End Time (hh:mm)',354 'Carpark Code',355 'Postal Code',356 'Total energy transferred by all EV charge points in the carpark (kWh)'357 ]358 359 for col_idx, header_text in enumerate(base_headers):360 worksheet.merge_range(0, col_idx, 1, col_idx, header_text, header_format)361 362 # Energy columns: Merge across all power rating columns in row 0363 if len(power_ratings) > 1:364 # Multiple power ratings - merge cells365 start_col = 6366 end_col = 6 + len(power_ratings) - 1367 worksheet.merge_range(0, start_col, 0, end_col, 368 'Average energy transferred (kWh) by each charge point with the same power rating', 369 header_format)370 371 # Write power ratings in row 1372 for i, power in enumerate(power_ratings):373 col_idx = 6 + i374 worksheet.write(1, col_idx, f'{power}kW', header_format)375 elif len(power_ratings) == 1:376 # Single power rating - write header in row 0 and power in row 1377 worksheet.write(0, 6, 378 'Average energy transferred (kWh) by each charge point with the same power rating', 379 header_format)380 worksheet.write(1, 6, f'{power_ratings[0]}kW', header_format)381 382 # Utilization columns: Merge across all power rating columns in row 0383 if len(power_ratings) > 1:384 # Multiple power ratings - merge cells385 start_col = 6 + len(power_ratings)386 end_col = 6 + 2 * len(power_ratings) - 1387 worksheet.merge_range(0, start_col, 0, end_col, 388 'Average time utilisation in percentage (%) of charge points per power rating\n(For non-publicly accessible charge points only)', 389 header_format)390 391 # Write power ratings in row 1392 for i, power in enumerate(power_ratings):393 col_idx = 6 + len(power_ratings) + i394 worksheet.write(1, col_idx, f'{power}kW', header_format)395 elif len(power_ratings) == 1:396 # Single power rating - write header in row 0 and power in row 1397 worksheet.write(0, 7, 398 'Average time utilisation in percentage (%) of charge points per power rating\n(For non-publicly accessible charge points only)', 399 header_format)400 worksheet.write(1, 7, f'{power_ratings[0]}kW', header_format)401 402 # Write data cells with borders (starting from row 2)403 for row in range(num_rows):404 for col in range(num_cols):405 cell_value = df_cp.iloc[row, col]406 407 # Determine which format to use based on column type408 if col == 5: # Total energy column409 cell_format = energy_format_border410 elif col >= 6 and col < 6 + len(power_ratings): # Energy columns411 cell_format = energy_format_border412 elif col >= 6 + len(power_ratings): # Utilization columns413 cell_format = percent_format_border414 else: # Other columns415 cell_format = border_format416 417 worksheet.write(row + 2, col, cell_value, cell_format)418 419 return output_file420 421 422def gradio_interface(file_a, start_date, end_date):423 # Create a temporary file for the output424 with tempfile.NamedTemporaryFile(delete=False, suffix='.xlsx') as tmp:425 output_file = tmp.name426 427 # Process the uploaded files and dates428 result = process_charging_sessions(file_a.name, output_file, start_date, end_date)429 430 # Return the path to the output file for download431 return result432 433 434# Define Gradio interface435iface = gr.Interface(436 fn=gradio_interface,437 inputs=[438 gr.File(label="Upload raw transactions CSV", file_types=[".csv"]),439 gr.Textbox(label="Start Date (YYYY-MM-DD)", value="2025-06-01"),440 gr.Textbox(label="End Date (YYYY-MM-DD)", value="2025-07-01")441 ],442 outputs=gr.File(label="Download output.xlsx"),443 title="Plugit Charging Transactions Convertor V2",444 description="Upload the raw Plugit transactions CSV file and specify the date range to process charging sessions and download the output Excel file."445)446 447 448# Launch the interface449if __name__ == "__main__":450 iface.launch()