一、项目结构zhi_xin_ban_cao/ │ ├── models/ │ ├── entities.py # 实体定义BBU / Board / Slot / Cell │ └── database.py # SQLAlchemy ORM SQLite │ ├── init_db.py # 初始化数据库 │ ├── problem/ │ ├── loader.py # 加载问题实例 │ └── state.py # 拓扑状态封装 │ ├── core/ │ ├── chromosome.py # GA 染色体 修复 变异 │ ├── cost.py # 成本函数0.7×距离 0.3×操作 │ ├── ga.py # GA 主循环 │ └── dqn.py # DQN Agent只引导方向 │ ├── utils/ │ └── distance.py # haversine │ ├── config.py # 全局参数 ├── main.py # 程序入口 └── requirements.txt二、模块2.1 config文件# config.py import os DB_PATH os.path.join(os.path.dirname(__file__), bbu_topology.db) POP_SIZE 50 MAX_GEN 200 MUTATION_RATE 0.15 DQN_STATE_DIM 32 DQN_ACTION_DIM 4 DQN_LR 1e-32.2 distance工具函数# utils/distance.py import math def haversine(lat1: float, lon1: float, lat2: float, lon2: float) - float: 球面距离米 R 6371000.0 phi1 math.radians(lat1) phi2 math.radians(lat2) dphi math.radians(lat2 - lat1) dlam math.radians(lon2 - lon1) a math.sin(dphi / 2) ** 2 \ math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))2.3数据库与实体2.3.1 entities# models/entities.py from dataclasses import dataclass dataclass class BBU: id: int name: str lat: float lon: float dataclass class Board: id: int bbu_id: int model: str # UBBPD2 / UBBPD4 property def capacity(self) - int: return 3 if self.model UBBPD2 else 6 dataclass class Slot: id: int board_id: int index: int cell_id: int | None None dataclass class Cell: id: int name: str is_new: bool lat: float lon: float2.3.2 databaseSQLite ORM# models/database.py from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, ForeignKey from sqlalchemy.orm import declarative_base, relationship, sessionmaker from config import DB_PATH Base declarative_base() class BBUTable(Base): __tablename__ bbu id Column(Integer, primary_keyTrue) name Column(String) lat Column(Float) lon Column(Float) class BoardTable(Base): __tablename__ board id Column(Integer, primary_keyTrue) bbu_id Column(Integer, ForeignKey(bbu.id)) model Column(String) class SlotTable(Base): __tablename__ slot id Column(Integer, primary_keyTrue) board_id Column(Integer, ForeignKey(board.id)) index Column(Integer) cell_id Column(Integer, nullableTrue) class CellTable(Base): __tablename__ cell id Column(Integer, primary_keyTrue) name Column(String) is_new Column(Boolean) lat Column(Float) lon Column(Float) engine create_engine(fsqlite:///{DB_PATH}, echoFalse) SessionLocal sessionmaker(bindengine) def init_db(): Base.metadata.create_all(bindengine)2.4 init_db 初始化数据库# init_db.py from models.database import engine, SessionLocal, Base from models.database import BBUTable, BoardTable, SlotTable, CellTable # 1. 创建表 Base.metadata.drop_all(bindengine) Base.metadata.create_all(bindengine) session SessionLocal() # 2. BBU5 台 bbus [ (BBU-1, 36.0611, 103.8340), (BBU-2, 36.0680, 103.8400), (BBU-3, 36.0550, 103.8500), (BBU-4, 36.0720, 103.8280), (BBU-5, 36.0620, 103.8700), ] bbu_ids {} for name, lat, lon in bbus: bbu BBUTable(namename, latlat, lonlon) session.add(bbu) session.flush() bbu_ids[name] bbu.id # 3. 板件每 BBU 6 插槽插 2~3 块板高占用 board_layout { BBU-1: [UBBPD4, UBBPD2, None, None, None, None], BBU-2: [UBBPD4, UBBPD2, UBBPD2, None, None, None], BBU-3: [UBBPD4, UBBPD2, None, None, None, None], BBU-4: [UBBPD4, UBBPD2, UBBPD2, None, None, None], BBU-5: [UBBPD4, UBBPD2, None, None, None, None], } boards [] for bbu_name, models in board_layout.items(): for model in models: if model is None: continue board BoardTable(bbu_idbbu_ids[bbu_name], modelmodel) session.add(board) session.flush() boards.append((board.id, model)) # 4. 槽位 slots [] for board_id, model in boards: cap 6 if model UBBPD4 else 3 for i in range(cap): slot SlotTable(board_idboard_id, indexi) session.add(slot) session.flush() slots.append(slot.id) # 5. 老小区占满大部分槽位 old_cell_id 1 for slot_id in slots[:-4]: # 故意留 4 个空槽 cell CellTable( namefOLD-{old_cell_id}, is_newFalse, lat36.06, lon103.84 ) session.add(cell) session.flush() session.query(SlotTable).filter_by(idslot_id).update({cell_id: cell.id}) old_cell_id 1 # 6. 新小区3 个 new_cells [ (NEW-1, 36.0613, 103.8342), (NEW-2, 36.0723, 103.8283), (NEW-3, 36.0623, 103.8703), ] for name, lat, lon in new_cells: session.add(CellTable(namename, is_newTrue, latlat, lonlon)) session.commit() print(✅ 数据库初始化完成5 BBU高占用率)2.5 问题加载2.5.1 loader# problem/loader.py from sqlalchemy import select from models.database import SessionLocal from models.database import BBUTable, BoardTable, SlotTable, CellTable from models.entities import BBU, Board, Slot, Cell def load_problem(): session SessionLocal() bbus { r.id: BBU(r.id, r.name, r.lat, r.lon) for r in session.execute(select(BBUTable)).scalars() } boards { r.id: Board(r.id, r.bbu_id, r.model) for r in session.execute(select(BoardTable)).scalars() } slots { r.id: Slot(r.id, r.board_id, r.index, r.cell_id) for r in session.execute(select(SlotTable)).scalars() } cells { r.id: Cell(r.id, r.name, r.is_new, r.lat, r.lon) for r in session.execute(select(CellTable)).scalars() } return bbus, boards, slots, cells2.5.2 state# problem/state.py from typing import Dict from models.entities import BBU, Board, Slot, Cell class TopologyState: def __init__( self, bbus: Dict[int, BBU], boards: Dict[int, Board], slots: Dict[int, Slot], cells: Dict[int, Cell] ): self.bbus bbus self.boards boards self.slots slots self.cells cells2.6 chromosome 染色体# core/chromosome.py import copy import random from typing import Dict from models.entities import Board, Slot class Chromosome: 染色体 板件位置 小区挂载 def __init__( self, board_placement: Dict[int, int], # board_id - bbu_id cell_placement: Dict[int, int] # cell_id - slot_id ): self.board_placement board_placement self.cell_placement cell_placement def clone(self): return Chromosome( copy.deepcopy(self.board_placement), copy.deepcopy(self.cell_placement) ) def is_valid(self, boards, slots): used_slots set() for cid, sid in self.cell_placement.items(): if sid in used_slots: return False used_slots.add(sid) load {} for cid, sid in self.cell_placement.items(): board slots[sid].board_id load[board] load.get(board, 0) 1 for bid, cnt in load.items(): if cnt boards[bid].capacity: return False return True2.7 cost 成本函数# core/cost.py from utils.distance import haversine from models.entities import BBU, Cell def calc_cost(chrom, state): dist 0.0 ops 0 for cid, sid in chrom.cell_placement.items(): cell state.cells[cid] if not cell.is_new: continue slot state.slots[sid] board state.boards[slot.board_id] bbu state.bbus[board.bbu_id] dist haversine(cell.lat, cell.lon, bbu.lat, bbu.lon) ops 1 total 0.7 * dist 0.3 * ops return total2.8 DQN# core/dqn.py import torch import torch.nn as nn import random class DQNAgent: def __init__(self, state_dim, action_dim, lr1e-3): self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, action_dim) ) self.opt torch.optim.Adam(self.net.parameters(), lrlr) def act(self, state, epsilon0.1): if random.random() epsilon: return random.randint(0, 3) with torch.no_grad(): return self.net(state).argmax().item()2.9 GA# core/ga.py import random from .chromosome import Chromosome class GA: def run(self, state): # 初始化种群 pop [self.random_chrom(state) for _ in range(50)] for gen in range(200): # 评估 # 选择 # 交叉 # 变异 pass def random_chrom(self, state): bp {bid: b.bbu_id for bid, b in state.boards.items()} cp {} for cid, cell in state.cells.items(): if cell.is_new: cp[cid] random.choice(list(state.slots.keys())) return Chromosome(bp, cp)2.10 main# main.py from problem.loader import load_problem from problem.state import TopologyState from core.ga import GA def main(): bbus, boards, slots, cells load_problem() state TopologyState(bbus, boards, slots, cells) ga GA() best ga.run(state) print(Best solution:, best) if __name__ __main__: main()三、场景补充chromosome ga 的业务逻辑未完成GA 变异算子​ 没有精确映射到1 / 2 / 3-A / 3-BChromosome.repair()​ 需要写成严格匹配三类场景