BTL Path Planning Simulation Overview and Key Functions | Môn Phân tích thiết kế hệ thống - Trường Đại học Bách khoa Hà Nội

lOMoARcPSD| 45315597 Nội dung
main.py............................................................................................................................................ 1
config.py........................................................................................................................................... 2
connect.py....................................................................................................................................... 3
filter.py............................................................................................................................................. 4
nx_solution.py................................................................................................................................. 6
raw_shortest_path.py...................................................................................................................... 8
uniTestGraphProcessor.py............................................................................................................... 9 main.py 1. Làm về cái gì? •
Entry point cho toàn bộ hệ thống mô phỏng lập kế hoạch đường đi (path planning simulation). •
Chạy lặp nhiều lần với các cấu hình khác nhau (solver, test automation, time
measurement), schedule và chạy event-driven simulation, rồi log kết quả vào CSV. 2. Có cái nào hay? •
Luân phiên giữa ba phương pháp giải:
1. LINK II solver (tự viết) 2. Parallel network-simplex 3. NetworkX •
Dùng discrevpy.simulator để schedule và chạy các Event một cách event-driven. lOMoARcPSD| 45315597 •
Quản lý cleanup thư mục output tự động qua DirectoryManager trước mỗi lần chạy,
giúp tách biệt dữ liệu giữa các thử nghiệm.
3. Làm như thế nào?
1. Khởi tạo o Import các module: Graph, AGV, Event,
Logger, GraphProcessor, v.v. o Dọn sạch thư mục output với dm.full_cleanup().
2. Định nghĩa các hàm helper o get_os(): phát hiện hệ điều hành.
o choose_solver(), choose_time_measurement(), choose_test_automation():
tương tác CLI để thiết lập config. 3.
Vòng lặp thử nghiệm (while config.count < 6) o
Tăng config.count, nếu cần thì dm.half_cleanup(). o
Chọn solver, chế độ test, mức độ đo thời gian.
o Tạo GraphProcessor, gọi use_in_main(), khởi tạo Graph và Event (qua
init_agvs_n_events, init_tasks, init_nodes_n_edges). o Sắp xếp và schedule sự
kiện, chạy simulator.run(). o Tính và in thời gian thực thi, ghi log vào file CSV.
o Gọi reset() để đưa simulator và AGV về trạng thái ban đầu. 4. Gồm có hàm gì? • get_os() • choose_solver() •
choose_time_measurement() •
choose_test_automation() •
schedule_events(events) • reset(simulator) •
Block if __name__ == "__main__": để thực thi cuối cùng. 5. Thể loại? lOMoARcPSD| 45315597 •
Script orchestration: CLI-driven, event-driven simulation driver. •
Kết hợp procedural (vòng lặp, gọi hàm) và interactive (nhập lựa chọn qua input()). config.py 1. Làm về cái gì?
config.py định nghĩa toàn bộ biến toàn cục dùng xuyên suốt hệ thống mô phỏng: •
Lựa chọn solver (solver_choice) •
Danh sách node bắt đầu (started_nodes) và các ID liên quan (ID) •
Tham số đồ thị (H, d) •
Số lượng AGV (num_max_agvs, numOfAGVs) và counters cho quá trình chạy (count,
reachingTargetAGVs, haltingAGVs) •
Theo dõi chi phí (totalCost) và thời gian giải (timeSolving) •
Chế độ mô phỏng (level_of_simulation – 0/1/2) •
Cờ test tự động (test_automation) và hiển thị đồ họa (draw) •
File chứa các hàm phụ trợ (functions_file) • … 2. Có cái nào hay? •
Tập trung cấu hình: gom mọi tham số ở một nơi, giúp dễ điều chỉnh mà không phải mò từng module. •
level_of_simulation được comment rõ 3 chế độ: “Fully Random”, “Random in the list”,
“SFM” – linh hoạt cho nhiều kiểu chạy thử. •
solver_choice mặc định là 'network-simplex', dễ chuyển đổi giữa các phương pháp giải
bằng cách sửa đúng biến này.
3. Làm như thế nào? •
Các module khác import config và truy cập trực tiếp các biến: lOMoARcPSD| 45315597 •
Không có logic xử lý trong file, chỉ khai báo và gán giá trị ban đầu cho biến. 4. Gồm có hàm gì? •
Không có hàm nào: toàn bộ là biến toàn cục và một dòng comment cho level_of_simulation. 5. Thể loại? •
Đây là configuration module (script cấu hình) – dùng để quản lý tham số cho toàn bộ
ứng dụng, tách biệt khỏi phần logic xử lý. connect.py 1. Làm về cái gì?
Module này cung cấp hàm tiện ích để: •
Thực thi các lệnh shell từ Python và lấy về kết quả đầu ra (stdout) Trích xuất các
giá trị thời gian (số nguyên) từ chuỗi đầu ra của lệnh đó 2. Có cái nào hay? •
subprocess.run với check=True: nếu lệnh trả về mã lỗi khác 0 sẽ tự động ném
CalledProcessError, giúp bắt lỗi rõ ràng. •
Tùy chọn capture_output linh hoạt: bật/tắt việc thu thập stdout theo nhu cầu. •
Dùng re.finditer với pattern r's (\d+)' để quét và chuyển từng nhóm số thành int, thu
được danh sách thời gian
3. Làm như thế nào? •
run_command(command, shell=False, capture_output=True)
1. Gọi subprocess.run(...) với text=True để trả về chuỗi.
2. Nếu capture_output=True, trả về result.stdout.strip(), ngược lại trả về None.
3. Bắt CalledProcessError và in lỗi rồi trả về None. •
extract_time_values(output)
1. Khởi tạo danh sách rỗng s_values.
2. Duyệt matches = re.finditer(r's (\d+)', output) để tìm tất cả các nhóm số sau ký tự "s ". lOMoARcPSD| 45315597
3. Với mỗi match, chuyển match.group(1) thành int và thêm vào s_values. 4. Trả về s_values 4. Gồm có hàm gì? •
run_command(command, shell=False, capture_output=True) •
extract_time_values(output) 5. Thể loại? •
Đây là utility module (hỗ trợ hệ thống) dành cho tương tác với shell và xử lý chuỗi đầu
ra, không chứa logic nghiệp vụ chính. filter.py 1. Làm về cái gì?
File này xử lý và lọc dữ liệu từ các file text để chuẩn bị input cho mô phỏng: •
remove_zero_lines loại bỏ các dòng không cần thiết (chỉ giữ dòng bắt đầu bằng 'c'/'s'
hoặc kết thúc bằng số > 0). •
filter_lines đọc điều kiện từ file sequence (seq-f.txt), quét file input, cộng dồn giá trị cuối
dòng và in log chi tiết quá trình lọc. 2. Có cái nào hay? •
Kết hợp lọc sơ bộ và tổng hợp có điều kiện, giúp chuẩn hóa dữ liệu trước khi chạy mô phỏng. •
In ra từng bước cập nhật biến count với cú pháp rõ ràng: old + inc = new, rất hữu ích cho việc debug.
3. Làm như thế nào? •
remove_zero_lines(file_path)
1. Đọc toàn bộ dòng từ file_path.
2. Giữ lại dòng nếu nó bắt đầu bằng 'c' hoặc 's', hoặc dòng có ký tự cuối cùng là số và giá trị > 0.
3. Ghi các dòng được giữ vào file filtered.txt. •
filter_lines(seq_file_path, input_file_path, H)
1. Đọc file sequence, với mỗi dòng bắt đầu 'f' trích ra hai số tiếp theo thành phần tử của mảng X. lOMoARcPSD| 45315597
2. Đọc toàn bộ file input, và với mỗi element trong X, tìm pattern "a " + element trong từng dòng.
3. Khi tìm thấy, lấy số cuối dòng (inc), tăng count += inc, in log chi tiết.
4. Cuối cùng in tổng count và gọi remove_zero_lines('seq-f.txt') để cập nhật file đã lọc. 4. Gồm có hàm gì? • remove_zero_lines(file_path) •
filter_lines(seq_file_path, input_file_path, H) 5. Thể loại? •
Đây là data preprocessing script: chuyên về lọc và tổng hợp dữ liệu trên các file text,
chuẩn bị input cho các bước mô phỏng đường đi. nx_solution.py 1. Làm về cái gì?
File này là một script CLI đơn giản để đọc dữ liệu đồ thị định dạng DIMACS, xây dựng đồ thị có
hướng với thuộc tính demand trên các node và capacity/weight trên các cạnh, rồi giải bài toán
Minimum Cost Flow bằng hàm network_simplex() của NetworkX và in kết quả ra console. 2. Có cái nào hay? •
Tận dụng network_simplex(): không phải tự triển khai thuật toán luồng chi phí thấp
nhất mà dùng giải thuật đã được tối ưu sẵn trong NetworkX. •
Xử lý DIMACS: parse dòng bắt đầu bằng 'n' để lấy demand, dòng 'a' để lấy capacity và
cost (lưu vào weight), phù hợp chuẩn DIMACS. •
Chế độ debug: có chỗ #pdb.set_trace() để dễ bật gỡ lỗi từng bước khi cần.
3. Làm như thế nào?
1. Nhập thư viện: import networkx as nx import pdb
2. Định nghĩa read_dimac_file(file_path): lOMoARcPSD| 45315597
o Khởi tạo G = nx.DiGraph().
o Mở file, đọc từng dòng, split() để lấy parts. o Nếu parts[0] == 'n': ID = parts[1] demand = -int(parts[2])
G.add_node(ID, demand=demand) o Nếu parts[0] == 'a': ID1, ID2 = parts[1], parts[2] U = int(parts[4]) (capacity) C = int(parts[5]) (cost)
G.add_edge(ID1, ID2, weight=C, capacity=U) 3. Giải luồng:
flowCost, flowDict = nx.network_simplex(G) 4. In kết quả: print("flowCost:", flowCost) print("flowDict:", flowDict) 5.
Phần chạy chính: file_path = input("Path to DIMAC file:") read_dimac_file(file_path)
``` :contentReference[oaicite:2]{index=2} 4. Gồm có hàm gì? •
read_dimac_file(file_path): duy nhất, đảm nhiệm từ việc parse file DIMACS đến gọi solver và in kết quả. •
Các hàm của NetworkX (nx.DiGraph(), nx.network_simplex()) được gọi bên trong nhưng
không được đóng gói thêm. 5. Thể loại? lOMoARcPSD| 45315597 •
Utility script / CLI demo:
o Không phải module importable, mà chạy trực tiếp qua python nx_solution.py.
o Phục vụ minh họa hoặc kiểm thử nhanh thuật toán Minimum Cost Flow với dữ liệu DIMACS raw_shortest_path.py 1. Làm về cái gì?
File này tính đường đi ngắn nhất giữa các điểm bắt đầu và điểm kết thúc trên một đồ thị có
hướng, được mô tả trong file đầu vào (định dạng tương tự DIMACS). Kết quả là độ dài ngắn
nhất từ mỗi start đến mỗi end, hoặc inf nếu không kết nối được. 2. Có cái nào hay? •
Tự triển khai Dijkstra với heapq mà không cần thư viện ngoài, giúp hiểu sâu cách hoạt
động của priority queue. •
Cơ chế đánh dấu start/end linh hoạt: đọc dòng n … 1 thành start, n … -1 thành end. •
Trả về float('inf') cho các cặp không kết nối, thuận tiện khi xử lý kết quả.
3. Làm như thế nào?
1. Đọc input (read_input(file_path)):
o Mở file, tách từng dòng theo dấu cách.
o Với dòng bắt đầu 'n' và số cuối là 1 → thêm vào start_points; nếu là -1 → thêm vào end_points.
o Với dòng 'a' → trích (source, destination, weight) vào edges.
2. Xây adjacency list trong find_shortest_paths:
o Duyệt edges, nhóm theo source thành dict graph[source] = [(dest, weight), …].
3. Dijkstra (dijkstra(graph, start)):
o Khởi distances[start]=0, push (0, start) vào min-heap. lOMoARcPSD| 45315597
o Lặp: pop node có current_distance nhỏ nhất, cập nhật neighbor nếu tìm được
đường ngắn hơn, push vào heap.
4. Tính kết quả (find_shortest_paths):
o Với mỗi start gọi dijkstra, sau đó xây dict {end: distances.get(end, inf)}.
5. In ra console (main):
o Hiển thị từng cặp Start point, End point và độ dài ngắn nhất. 4. Gồm có hàm gì? Hàm Mô tả
read_input(file_path) Đọc file, phân loại start_points, end_points, và edges. dijkstra(graph,
start) Tính khoảng cách ngắn nhất từ start đến mọi node với min-heap.
find_shortest_paths(...) Xây graph, gọi dijkstra cho mỗi start, trả kết quả dạng dict. main(file_path)
Điều phối: gọi read_input → find_shortest_paths → in kết quả. 5. Thể loại? •
Script CLI thuật toán: chạy trực tiếp qua python raw_shortest_path.py, dùng để
benchmark hoặc debug tính đúng đắn của Dijkstra trên input file. •
Phù hợp làm baseline so sánh với các giải thuật phức tạp hơn (như NetworkX hay network simplex). uniTestGraphProcessor.py 1. Làm về cái gì? •
Đây là bộ kiểm thử (unit tests) dành cho lớp GraphProcessor, đảm bảo các bước xử lý
đồ thị (như nạp file, tạo ma trận, thêm constraints, khởi tạo node/edge, cập nhật graph)
hoạt động đúng theo yêu cầu lOMoARcPSD| 45315597 2. Có cái nào hay? •
Sử dụng nhiều hàm assert_… để kiểm tra tính toàn vẹn của graph ở các giai đoạn khác nhau:
o Số lượng node/edge phải đúng sau mỗi bước
o Node và edge thỏa mãn ràng buộc thời gian (TimeWindowNode,
RestrictionNode, TimeWindowEdge, RestrictionEdge) •
Kết hợp in thông báo rõ ràng (“Test cho file 2ndSimple.txt”) và tự động lặp qua hai tình
huống earliness/tardiness để kiểm thử cả hai kịch bản.
3. Làm như thế nào?
1. Import và khởi tạo
o Nhập các lớp từ model và controller (ví dụ Graph, Event, TimeWindowNode, RestrictionEdge, v.v.)
o Tạo instance processor = GraphProcessor() và tắt in debug (processor.print_out = False).
2. Nạp dữ liệu & cấu hình
o Dùng file "2ndSimple.txt", set started_nodes, H, d, alpha, beta, gamma, restrictions, v.v.
3. Chạy pipeline xử lý
o processor.process_input_file() → generate_hm_matrix() →
generate_adj_matrix() → create_tsg_file() o Lặp hai lần để test cả earliness và
tardiness, gọi add_time_windows_constraints()
o Thiết lập restriction_count, start_ban, end_ban, restrictions, rồi process_restrictions()
4. Khởi tạo Graph & update o Tạo graph = Graph(processor) o Gọi
processor.init_nodes_n_edges(graph) để thêm node/edge vào graph o Kiểm tra tổng số
edge/node khớp với processor.tsedges/processor.ts_nodes
o Gọi processor.update_graph(id1, id2, c12) để mô phỏng cập nhật graph sau sự kiện lOMoARcPSD| 45315597
5. Kiểm thử các ràng buộc o Gọi tuần tự 8 hàm assert_… để xác minh:
1. assert_Nodes(graph, current_time)
2. assert_Edges(graph, current_time)
3. assert_connection_TimeWindowNodes(graph)
4. assert_RestrictionNodes(graph)
5. So sánh số TimeWindowNode đúng = 2
6. assert_number_TimeWindowEdges(graph, current_time)
7. assert_new_RestrictionNodes(graph) 8.
assert_numberOf_RestrictionNodes(graph) 4. Gồm có hàm gì? •
assert_Nodes(graph, current_time): Đếm node thỏa mãn time ≥ current_time (loại trừ TimeWindow/RestrictionNode). •
assert_Edges(graph, current_time): Kiểm tra các edge (ngoại trừ
TimeWindow/RestrictionEdge) có nguồn thỏa time ≥ current_time. •
assert_connection_TimeWindowNodes(graph): Đảm bảo mỗi TimeWindowNode có ít
nhất một edge nối vào. •
assert_number_TimeWindowEdges(graph, current_time): So sánh số
TimeWindowEdge mới đúng bằng cũ trừ các edge không thỏa ràng buộc thời gian. •
assert_RestrictionNodes(graph): Mỗi RestrictionNode phải có ít nhất một edge nối vào. •
assert_new_RestrictionNodes(graph): Tương tự với RestrictionEdge có nguồn không phải RestrictionNode. •
assert_numberOf_RestrictionNodes(graph): Kiểm tra số RestrictionEdge với đích không
phải RestrictionNode sau lọc theo thời gian. •
Block kiểm thử cuối file: Tạo processor, thiết lập tham số, chạy pipeline và gọi các hàm assert. 5. Thể loại? lOMoARcPSD| 45315597 •
Unit Test Module: Dùng để xác minh tính đúng đắn của lớp GraphProcessor sau mỗi
bước xử lý, đảm bảo các ràng buộc về thời gian và cấu trúc graph luôn được duy trì. •
Phù hợp để tích hợp CI/CD, phát hiện lỗi ngay khi có thay đổi code.