233 lines
8.0 KiB
Python
233 lines
8.0 KiB
Python
"""
|
|
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()
|