Как оптимизировать моделирование парковочного места, чтобы обрабатывать больше автомобилей без дублирования ⇐ Python

[Anonymous]

У меня есть программа моделирования парковочных мест для проверки кода счетчика парковки, но в настоящее время она ограничена моделированием только 4 автомобилей. Основная проблема заключается в избыточности функций Floor_wait, каждая из которых обрабатывает только один этаж. Я хочу масштабировать симуляцию, чтобы эффективно обрабатывать больше автомобилей, не дублируя код для каждого этажа.

Код: Выделить всё

# Import library
import json
import threading
import queue
import time
import random
from tkinter import *
from tkinter import ttk

#try catch block

# need to make more queues if you want more floors (the Wq one)
iq = queue.Queue()
wq1 = queue.Queue()
wq2 = queue.Queue()
wq3 = queue.Queue()
oq = queue.Queue()
lock = threading.Lock()
TEMP_FILE = 'parking_slot.json'  # File path

# Load or initialize parking state
try:
with open(TEMP_FILE, 'r') as f:
data = json.load(f)
total_slots = data.get('total_slots', 300)
floor_slots = data.get('floor_slots', {'Floor 1': 100, 'Floor 2': 100, 'Floor 3': 100})
except FileNotFoundError:
total_slots = 300
floor_slots = {'Floor 1': 100, 'Floor 2': 100, 'Floor 3': 100}

# For the windows display stuff
root = Tk()
frm = ttk.Frame(root, padding=10)
total_label = ttk.Label(frm, text=f"Total slots: {total_slots}")
floor_labels = {}

# code not used juut here if i want to update it with a saving feature
def save_data():
data = {
'total_slots': total_slots,
'floor_slots': floor_slots
}
with open(TEMP_FILE, 'w') as temp_file:
json.dump(data, temp_file)

def stop_threads():
iq.put(None)  # Signal to floor_input to stop
wq1.put(None)  # Signal to floor1_wait to stop
wq2.put(None)  # Signal to floor2_wait to stop
wq3.put(None)  # Signal to floor3_wait to stop
oq.put(None)  # Signal to main_output to stop

def window_display():  # code to display the slots
global total_slots, floor_slots, floor_labels, root, total_label, frm
root.title("Car Park Display Counter Simulation")
frm.grid()
total_label.grid(column=0, row=0)
for i, (floor, slots) in enumerate(floor_slots.items()):
floor_label = ttk.Label(frm, text=f"{floor}: {slots} slots")
floor_label.grid(column=0, row=i + 1)
floor_labels[floor] = floor_label

root.protocol("WM_DELETE_WINDOW", stop_threads)

def update_data():  # Code to update the window that displays the slots
global total_slots, floor_slots, floor_labels, root, total_label
root.after(0, lambda: total_label.config(text=f"Total slots: {total_slots}"))
for floor in floor_slots.keys():
root.after(0, lambda floor=floor: floor_labels[floor].config(text=f"{floor}: {floor_slots[floor]} slots"))

def main_input():
global total_slots
for i in range(100):  # Chhange this depending on how many cars u want to go in
if total_slots > 0:
car = f'car {random.randint(1, 100)}'  # Generate a unique car name
iq.put(car)
total_slots -= 1
update_data()
else:
time.sleep(1)  # Wait a bit if no slots are available

time.sleep(random.uniform(1, 3))  # Simulates human delay

if not iq.empty() and iq.queue[0] is None:
break

def floor_input():  # Code that decides on what floor the car will go to
global floor_slots
while True:
try:
car = iq.get(timeout=1)
except queue.Empty:
continue

if car is None:
break
floor_name = random.choice(list(floor_slots.keys()))
with lock:  # A lock to prevent race conditions
if floor_slots[floor_name] >  0:  # Check if there are available slots on the floor
floor_slots[floor_name] -= 1
wait_time = {
'Floor 1': 1,
'Floor 2': 2,
'Floor 3': 3,
}.get(floor_name, 1)
time.sleep(wait_time)

# Direct the car to the appropriate queue based on the floor
if floor_name == 'Floor 1':
wq1.put(car)
elif floor_name == 'Floor 2':
wq2.put(car)
elif floor_name == 'Floor 3':
wq3.put(car)
else:
iq.put(car)  # If no slots are available, put the car back in the input queue

if not iq.empty() and iq.queue[0] is None:
break

update_data()

def floor1_wait():  # The code that simulates the car waiting in the parking space
global floor_slots
while True:
try:
car1 = wq1.get(timeout=1)
except queue.Empty:
continue
if car1 is None:
break
time.sleep(random.uniform(3, 8))  # Car waiting simulation
floor_slots['Floor 1'] += 1
oq.put(car1)
update_data()
time.sleep(1)  # Car going to the exit
if not wq1.empty() and wq1.queue[0] is None:
break

def floor2_wait():  # The code that simulates the car waiting in the parking space
global floor_slots
while True:
try:
car2 = wq2.get(timeout=1)
except queue.Empty:
continue
if car2 is None:
break
time.sleep(random.uniform(3, 8))
floor_slots['Floor 2'] += 1
oq.put(car2)
update_data()
time.sleep(2)
if not wq2.empty() and wq2.queue[0] is None:
break

def floor3_wait():  # The code that simulates the car waiting in the parking space
global floor_slots
while True:
try:
car3 = wq3.get(timeout=1)
except queue.Empty:
continue
if car3 is None:
break
time.sleep(random.uniform(3, 8))
floor_slots['Floor 3'] += 1
oq.put(car3)
update_data()
time.sleep(3)
if not wq3.empty() and wq3.queue[0] is None:
break

def main_output():  # Code that sorts out cars that are going out
global total_slots
while True:
try:
car = oq.get(timeout=1)
except queue.Empty:
continue

if car is None:
break
total_slots += 1
update_data()
time.sleep(random.uniform(1, 5))

while not iq.empty():
car = iq.get()
if car is not None:
total_slots += 1
update_data()
time.sleep(random.uniform(1, 5))

# Display the window
window_display()
update_data()

# Create threads
main_input_thread = threading.Thread(target=main_input)
floor_input_thread = threading.Thread(target=floor_input)
floor1_output_thread = threading.Thread(target=floor1_wait)
floor2_output_thread = threading.Thread(target=floor2_wait)
floor3_output_thread = threading.Thread(target=floor3_wait)
main_output_thread = threading.Thread(target=main_output)

# Start threads
main_input_thread.start()
floor_input_thread.start()
floor1_output_thread.start()
floor2_output_thread.start()
floor3_output_thread.start()
main_output_thread.start()

# Run the Tkinter main loop
root.mainloop()  # For the display window

# After the GUI is closed, wait for threads to finish
main_input_thread.join()
floor_input_thread.join()
floor1_output_thread.join()
floor2_output_thread.join()
floor3_output_thread.join()
main_output_thread.join()

print("Simulation Ended.")

На данный момент код работает нормально, но для моделирования более 4 автомобилей мне нужно избегать дублирования функции Floor_wait для каждого этажа. Как я могу реорганизовать свой код, чтобы эффективно обрабатывать больше автомобилей и этажей?
Будем очень признательны за любую помощь или предложения.

Подробнее здесь: https://stackoverflow.com/questions/792 ... uplication