""" run_all_paths.py - single threaded RAM batch version Loads batches of files into RAM, processes sequentially, frees memory. No worker processes = no memory multiplication. """ import pandas as pd import sys import os import re import csv import time import gc from datetime import datetime, timedelta # ── Configuration ──────────────────────────────────────────────────────────── ROSETTA_PATH = r'D:\ATT\ATT_Rosetta_populated.csv' GPD_PATH = r'D:\ATT\ATT_GPD_ALL.xlsx' ERROR_LOG = r'D:\ATT\error_log.csv' RUN_LOG = r'D:\ATT\run_log.csv' MAX_BATCH_GB = 10 # max GB of input files on disk per batch # ───────────────────────────────────────────────────────────────────────────── sys.path.insert(0, r'D:\ATT') os.environ['TQDM_DISABLE'] = '1' import kizer_krunch_code as kc def build_gpd_lookup(): gpd_df = pd.read_excel(GPD_PATH, dtype=str) lookup = {} for _, row in gpd_df.iterrows(): tx = str(row[kc.GENERALPATHDATA_COLUMNS["tx_call_sign"]]).strip() rx = str(row[kc.GENERALPATHDATA_COLUMNS["rx_call_sign"]]).strip() lookup[(tx, rx)] = { "channels": row[kc.GENERALPATHDATA_COLUMNS["channels"]], "bandwidth": row[kc.GENERALPATHDATA_COLUMNS["bandwidth"]], "frequency": row[kc.GENERALPATHDATA_COLUMNS["fr_frequency_assigned_MHz"]], "gain": row[kc.GENERALPATHDATA_COLUMNS["rxan_gain_dBi"]], "tilt": row[kc.GENERALPATHDATA_COLUMNS["RxTilt_deg"]], "bearing": row[kc.GENERALPATHDATA_COLUMNS["RxBearing_deg"]], "ant_model": re.sub(r'-\d{4}-\d{4}$', '', re.sub(r'-WC\.CSV$', '', str(row[kc.GENERALPATHDATA_COLUMNS["AltRcvAntenna"]]), flags=re.IGNORECASE)).strip(), "tx_location_name": str(row[kc.GENERALPATHDATA_COLUMNS["txan_location_name"]]).strip(), "rx_location_name": str(row[kc.GENERALPATHDATA_COLUMNS["rxan_location_name"]]).strip(), } return lookup def read_csv_rows(filepath): try: with open(filepath, mode='r') as f: return list(csv.reader(f)) except Exception: return None def make_batches(path_nums, rosetta_rows, a_key, b_key): """Split paths into batches based on actual file sizes on disk.""" max_bytes = MAX_BATCH_GB * 1024 ** 3 batches, current, current_bytes = [], [], 0 for pn in path_nums: row = rosetta_rows[pn] size = 0 for key in (a_key, b_key): try: size += os.path.getsize(str(row[kc.FNC_COLUMNS[key]]).strip()) except Exception: pass if current and current_bytes + size > max_bytes: batches.append(current) current, current_bytes = [], 0 current.append(pn) current_bytes += size if current: batches.append(current) return batches def fmt(seconds): return str(timedelta(seconds=int(seconds))) def save_logs(run_rows, error_rows): for _ in range(5): try: pd.DataFrame(run_rows).to_csv(RUN_LOG, index=False) if error_rows: pd.DataFrame(error_rows).to_csv(ERROR_LOG, index=False) return except PermissionError: time.sleep(1) def run_pass(label, path_nums, rosetta_rows, a_key, b_key, calc_fn, run_rows, error_rows, succeeded, failed, completed, total_tasks, wall_start, gpd_lookup=None): batches = make_batches(path_nums, rosetta_rows, a_key, b_key) print(f"\n--- {label} ({len(path_nums)} paths, {len(batches)} batches) ---") for b_num, batch in enumerate(batches, 1): print(f"\n Loading batch {b_num}/{len(batches)} ({len(batch)} paths)...") t0 = time.time() # Load this batch into RAM batch_data = {} for pn in batch: row = rosetta_rows[pn] a = read_csv_rows(str(row[kc.FNC_COLUMNS[a_key]]).strip()) b = read_csv_rows(str(row[kc.FNC_COLUMNS[b_key]]).strip()) batch_data[pn] = (a, b) print(f" Loaded in {fmt(time.time()-t0)}. Processing {len(batch)} paths...") # Process sequentially from RAM for pn in batch: row = rosetta_rows[pn] path_id = str(row['Path_ID']).strip() drive = str(row[kc.FNC_COLUMNS['drive']]).strip() folder = str(row[kc.FNC_COLUMNS['folder']]).strip().strip('\\') subfolder = str(row[kc.FNC_COLUMNS['subfolder']]).strip() a_data, b_data = batch_data[pn] completed += 1 path_start = time.time() path_errors = [] try: calc_fn(row, drive, folder, subfolder, a_data, b_data, path_errors, gpd_lookup) except Exception as e: path_errors.append(str(e)) elapsed = round(time.time() - path_start, 1) status = 'ERROR' if path_errors else 'OK' ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S') wall_elapsed = time.time() - wall_start rate = completed / wall_elapsed if wall_elapsed > 0 else 0 eta = fmt((total_tasks - completed) / rate) if rate > 0 else '?' pct = 100 * completed / total_tasks print(f"[{completed}/{total_tasks} {pct:.1f}%] " f"{'OK ' if status=='OK' else 'FAIL'} " f"{pn} {path_id} ({elapsed}s) ETA:{eta}") run_rows.append({'path_num': pn, 'path_id': path_id, 'status': status, 'elapsed_s': elapsed, 'errors': ' | '.join(path_errors), 'timestamp': ts}) if path_errors: failed += 1 for msg in path_errors: error_rows.append({'path_num': pn, 'path_id': path_id, 'error': msg, 'timestamp': ts}) else: succeeded += 1 if completed % 10 == 0: save_logs(run_rows, error_rows) # Free this batch del batch_data gc.collect() print(f" Batch {b_num} done. Elapsed: {fmt(time.time()-wall_start)}") return succeeded, failed, completed def interference_calc(row, drive, folder, subfolder, a_data, b_data, path_errors, gpd_lookup): AIntFunction = str(row[kc.FNC_COLUMNS['AIntFunction']]).strip() AIntInput = str(row[kc.FNC_COLUMNS['fileAInt']]).strip() BIntInput = str(row[kc.FNC_COLUMNS['fileBInt']]).strip() base_path = f"{drive}\\{folder}\\{subfolder}" AIntOutput = f"{base_path}\\{subfolder}_A_{AIntFunction}.csv" BIntOutput = f"{base_path}\\{subfolder}_B_{AIntFunction}.csv" for label, inp, out, data in [ ('A INT', AIntInput, AIntOutput, a_data), ('B INT', BIntInput, BIntOutput, b_data), ]: try: kc.run_interference_calculation(inp, out, drive, folder, gpd_lookup, preloaded_data=data) except Exception as e: path_errors.append(f"{label}: {e}") def terrain_calc(row, drive, folder, subfolder, a_data, b_data, path_errors, gpd_lookup): ATerrFunction = str(row[kc.FNC_COLUMNS['ATerrFunction']]).strip() ATerrInput = str(row[kc.FNC_COLUMNS['fileATerr']]).strip() BTerrInput = str(row[kc.FNC_COLUMNS['fileBTerr']]).strip() base_path = f"{drive}\\{folder}\\{subfolder}" ATerrOutput = f"{base_path}\\{subfolder}_A_{ATerrFunction}.csv" BTerrOutput = f"{base_path}\\{subfolder}_B_{ATerrFunction}.csv" for label, inp, out, data in [ ('A TER', ATerrInput, ATerrOutput, a_data), ('B TER', BTerrInput, BTerrOutput, b_data), ]: try: kc.run_terrain_calculation(inp, out, preloaded_data=data) except Exception as e: path_errors.append(f"{label}: {e}") def main(): wall_start = time.time() print(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"Single threaded | Max batch: {MAX_BATCH_GB}GB on disk") print("Loading Rosetta...") rosetta = pd.read_csv(ROSETTA_PATH) path_nums = rosetta['path_num'].astype(str).str.strip().tolist() total_paths = len(path_nums) total_tasks = total_paths * 2 print(f"Paths: {total_paths}") rosetta_rows = {} for pn in path_nums: idx = rosetta.index[rosetta['path_num'].astype(str).str.strip() == pn][0] rosetta_rows[pn] = rosetta.loc[idx].to_dict() print("Loading GPD...") gpd_lookup = build_gpd_lookup() print(f"GPD loaded: {len(gpd_lookup)} paths.") run_rows = [] error_rows = [] succeeded = 0 failed = 0 completed = 0 try: succeeded, failed, completed = run_pass( 'PASS 1: Interference', path_nums, rosetta_rows, 'fileAInt', 'fileBInt', interference_calc, run_rows, error_rows, succeeded, failed, completed, total_tasks, wall_start, gpd_lookup) print(f"\nInterference done. Elapsed: {fmt(time.time()-wall_start)}") succeeded, failed, completed = run_pass( 'PASS 2: Terrain', path_nums, rosetta_rows, 'fileATerr', 'fileBTerr', terrain_calc, run_rows, error_rows, succeeded, failed, completed, total_tasks, wall_start) print(f"\nTerrain done. Elapsed: {fmt(time.time()-wall_start)}") except KeyboardInterrupt: print("\nInterrupted — saving logs...") save_logs(run_rows, error_rows) print(f"\n{'='*60}") print(f"Complete: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"Total time: {fmt(time.time()-wall_start)}") print(f"Succeeded: {succeeded} Failed: {failed}") print(f"Run log: {RUN_LOG}") if failed: print(f"Error log: {ERROR_LOG}") if __name__ == '__main__': main()