""" Step 06: 约束满足检验 输入: 03_allocate.xlsx, 05_schedule.xlsx 输出: 06_validate.xlsx 功能: 1. 检验资源约束: Σk_i = 730 2. 检验覆盖约束: k_i >= 1 3. 检验日容量约束: 每日恰好2站点 4. 检验无重复约束: 同一天不重复访问同一站点 5. 检验排程一致性: 排程中的访问次数 = 分配的k_i """ import pandas as pd import numpy as np from pathlib import Path # 路径配置 ALLOCATE_PATH = Path(__file__).parent / "03_allocate.xlsx" SCHEDULE_PATH = Path(__file__).parent / "05_schedule.xlsx" OUTPUT_PATH = Path(__file__).parent / "06_validate.xlsx" # 约束参数 N_TOTAL = 730 MIN_VISITS = 1 DAILY_CAPACITY = 2 N_DAYS = 365 def main(): print("=" * 60) print("Step 06: 约束满足检验") print("=" * 60) # 1. 读取数据 print(f"\n[1] 读取输入数据") df_allocate = pd.read_excel(ALLOCATE_PATH) df_calendar = pd.read_excel(SCHEDULE_PATH, sheet_name='calendar') df_site_dates = pd.read_excel(SCHEDULE_PATH, sheet_name='site_dates') print(f" 分配数据: {len(df_allocate)} 站点") print(f" 日历数据: {len(df_calendar)} 天") # 存储检验结果 results = [] # 2. 检验资源约束: Σk_i = N_TOTAL print(f"\n[2] 检验资源约束: Σk_i = {N_TOTAL}") total_k = df_allocate['k'].sum() c1_pass = total_k == N_TOTAL results.append({ 'constraint': 'C1: 资源约束', 'formula': 'Σk_i = 730', 'expected': N_TOTAL, 'actual': total_k, 'passed': c1_pass, 'details': f"总访问次数 = {total_k}" }) print(f" Σk_i = {total_k}, 预期 = {N_TOTAL}") print(f" 结果: {'✅ 通过' if c1_pass else '❌ 失败'}") # 3. 检验覆盖约束: k_i >= MIN_VISITS print(f"\n[3] 检验覆盖约束: k_i >= {MIN_VISITS}") min_k = df_allocate['k'].min() sites_below_min = df_allocate[df_allocate['k'] < MIN_VISITS] c2_pass = len(sites_below_min) == 0 results.append({ 'constraint': 'C2: 覆盖约束', 'formula': f'k_i >= {MIN_VISITS}', 'expected': f'>= {MIN_VISITS}', 'actual': f'min = {min_k}', 'passed': c2_pass, 'details': f"最小访问次数 = {min_k}, 违反站点数 = {len(sites_below_min)}" }) print(f" min(k_i) = {min_k}, 预期 >= {MIN_VISITS}") print(f" 结果: {'✅ 通过' if c2_pass else '❌ 失败'}") # 4. 检验日容量约束: 每日恰好2站点 print(f"\n[4] 检验日容量约束: 每日恰好 {DAILY_CAPACITY} 站点") # 统计每天的站点数 daily_counts = [] for _, row in df_calendar.iterrows(): count = 0 if pd.notna(row['site_1_id']): count += 1 if pd.notna(row['site_2_id']): count += 1 daily_counts.append(count) df_calendar['site_count'] = daily_counts days_not_full = df_calendar[df_calendar['site_count'] != DAILY_CAPACITY] c3_pass = len(days_not_full) == 0 results.append({ 'constraint': 'C3: 日容量约束', 'formula': f'每日站点数 = {DAILY_CAPACITY}', 'expected': f'所有天 = {DAILY_CAPACITY}', 'actual': f'min={min(daily_counts)}, max={max(daily_counts)}', 'passed': c3_pass, 'details': f"不满足的天数 = {len(days_not_full)}" }) print(f" 每日站点数: min={min(daily_counts)}, max={max(daily_counts)}") print(f" 不满足的天数: {len(days_not_full)}") print(f" 结果: {'✅ 通过' if c3_pass else '❌ 失败'}") # 5. 检验无重复约束: 同一天不重复访问同一站点 print(f"\n[5] 检验无重复约束: 同一天不重复访问同一站点") duplicate_days = [] for _, row in df_calendar.iterrows(): if pd.notna(row['site_1_id']) and pd.notna(row['site_2_id']): if row['site_1_id'] == row['site_2_id']: duplicate_days.append(row['day']) c4_pass = len(duplicate_days) == 0 results.append({ 'constraint': 'C4: 无重复约束', 'formula': 'site_1 ≠ site_2 (同一天)', 'expected': '无重复', 'actual': f'重复天数 = {len(duplicate_days)}', 'passed': c4_pass, 'details': f"重复的天: {duplicate_days[:5]}..." if duplicate_days else "无" }) print(f" 重复访问的天数: {len(duplicate_days)}") print(f" 结果: {'✅ 通过' if c4_pass else '❌ 失败'}") # 6. 检验排程一致性: 排程中的访问次数 = 分配的k_i print(f"\n[6] 检验排程一致性: 排程访问次数 = 分配的k_i") # 从日历中统计每个站点的访问次数 schedule_counts = {} for _, row in df_calendar.iterrows(): for site_col in ['site_1_id', 'site_2_id']: site_id = row[site_col] if pd.notna(site_id): site_id = int(site_id) schedule_counts[site_id] = schedule_counts.get(site_id, 0) + 1 # 对比分配的k_i mismatch_sites = [] for _, row in df_allocate.iterrows(): site_id = row['site_id'] allocated_k = row['k'] scheduled_k = schedule_counts.get(site_id, 0) if allocated_k != scheduled_k: mismatch_sites.append({ 'site_id': site_id, 'allocated_k': allocated_k, 'scheduled_k': scheduled_k, 'diff': scheduled_k - allocated_k }) c5_pass = len(mismatch_sites) == 0 results.append({ 'constraint': 'C5: 排程一致性', 'formula': '排程次数 = k_i', 'expected': '所有站点一致', 'actual': f'不一致站点数 = {len(mismatch_sites)}', 'passed': c5_pass, 'details': str(mismatch_sites[:5]) if mismatch_sites else "全部一致" }) print(f" 不一致的站点数: {len(mismatch_sites)}") print(f" 结果: {'✅ 通过' if c5_pass else '❌ 失败'}") # 7. 检验总天数 print(f"\n[7] 检验总天数: {N_DAYS} 天") actual_days = len(df_calendar) c6_pass = actual_days == N_DAYS results.append({ 'constraint': 'C6: 总天数', 'formula': f'日历天数 = {N_DAYS}', 'expected': N_DAYS, 'actual': actual_days, 'passed': c6_pass, 'details': f"日历包含 {actual_days} 天" }) print(f" 日历天数 = {actual_days}, 预期 = {N_DAYS}") print(f" 结果: {'✅ 通过' if c6_pass else '❌ 失败'}") # 8. 汇总 print(f"\n" + "=" * 60) print("检验结果汇总") print("=" * 60) df_results = pd.DataFrame(results) all_pass = df_results['passed'].all() print(f"\n{'约束':<20} {'结果':<10} {'详情'}") print("-" * 60) for _, row in df_results.iterrows(): status = '✅ 通过' if row['passed'] else '❌ 失败' print(f"{row['constraint']:<20} {status:<10} {row['details']}") print("-" * 60) print(f"总体结果: {'✅ 所有约束通过' if all_pass else '❌ 存在约束违反'}") # 9. 保存输出 print(f"\n[8] 保存输出: {OUTPUT_PATH}") with pd.ExcelWriter(OUTPUT_PATH, engine='openpyxl') as writer: df_results.to_excel(writer, sheet_name='validation_results', index=False) # 保存详细的每日统计 df_calendar[['day', 'site_1_id', 'site_2_id', 'site_count']].to_excel( writer, sheet_name='daily_check', index=False ) # 保存站点一致性检查 consistency_data = [] for _, row in df_allocate.iterrows(): site_id = row['site_id'] consistency_data.append({ 'site_id': site_id, 'site_name': row['site_name'], 'allocated_k': row['k'], 'scheduled_k': schedule_counts.get(site_id, 0), 'match': row['k'] == schedule_counts.get(site_id, 0) }) pd.DataFrame(consistency_data).to_excel( writer, sheet_name='site_consistency', index=False ) print(f" 已保存3个工作表") print("\n" + "=" * 60) print("Step 06 完成") print("=" * 60) return df_results, all_pass if __name__ == "__main__": main()