import csv
import math
import random
from functools import partial
import numpy as np
from typing import List, Tuple, Union
from models.entities import move_to_another_block
from models.entities import best_block_candidates, wait_for_thrash_generation, return_to_landfill, Truck
from models.map_setup import create_random_list, create_map
from models.utils import DaysOfTheWeek, Blocks
from models.entities import create_trucks
if __name__ == '__main__':
NUMBER_OF_TRUCKS = 2
DAYS = 150
NUMBER_OF_BLOCKS = 30
truck_block_sequences = [[1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
[10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0],
[5, 4, 3, 2, 1, 6, 7, 8, 9, 10],
[6, 7, 8, 9, 10, 1, 2, 3, 4, 5]]
#trucks_pools = create_trucks(NUMBER_OF_TRUCKS,truck_block_sequences)
trucks_pools = [(Truck(name=f"truck {i + 1}", capacity_kg=6,
block_sequence=truck_block_sequences[i % len(truck_block_sequences)]), 0) for i in
range(NUMBER_OF_TRUCKS)]
the_route_map = create_map(NUMBER_OF_BLOCKS)
MEAN_THRASH_BLOCK = 72_000 / NUMBER_OF_BLOCKS
VARIANCE_THRASH = 0.25
TIME_THRASH_GENERATION = 12
# en func defines la funcion para generar los datos
time_landfill = create_random_list(NUMBER_OF_BLOCKS,
func=lambda: int(math.log(np.random.lognormal(45, 5))))
from_chorillo = create_random_list(NUMBER_OF_BLOCKS,
func=lambda: int(math.log(np.random.lognormal(7, 2))))
special_increments = {DaysOfTheWeek.friday: 0.25,
DaysOfTheWeek.saturday: 0.25,
DaysOfTheWeek.sunday: 0.25,
DaysOfTheWeek.monday: 0.25,
DaysOfTheWeek.tuesday:0.25,
DaysOfTheWeek.wednesday:0.25,
DaysOfTheWeek.thursday:0.25,}
history = []
thrash_state = []
trash_in_block = [0] * NUMBER_OF_BLOCKS
block_sequences_generator = partial(best_block_candidates,
truck_block_sequences,
time_landfill,
the_route_map,
from_chorillo,
NUMBER_OF_BLOCKS)
move_to = partial(move_to_another_block,
time_landfill,
from_chorillo,
the_route_map)
for day in range(DAYS):
std = MEAN_THRASH_BLOCK * VARIANCE_THRASH
temp = create_random_list(NUMBER_OF_BLOCKS,
lambda: abs(int(random.gauss(MEAN_THRASH_BLOCK, std))))
trash_in_block = [a + b for a, b in zip(trash_in_block, temp)]
global_time = 0
trash_generation = 1
while global_time < 16 * 60:
total_time_spend = 0
for current_truck in trucks_pools:
if sum(trash_in_block) == 0:
break
if current_truck.turn % 3 == 0:
continue
thrash_state.append([*trash_in_block])
block_sequences = block_sequences_generator(
current_truck,
trash_in_block
)
select_option = partial(move_to, trash_in_block, current_truck)
moved = any((history_move := select_option(next_block)) for next_block in block_sequences)
if not moved:
moved_landfill = return_to_landfill(current_truck, time_landfill) | {'day': DaysOfTheWeek(day % 7)}
history.append(moved_landfill)
else:
history.append(history_move | {'day': DaysOfTheWeek(day % 7)})
current_truck.history[-1] |= {'day': DaysOfTheWeek(day % 7)}
total_time_spend += history[-1].get('total_time_spend')
if sum(trash_in_block) == 0:
temp = wait_for_thrash_generation(trucks_pools,
abs(global_time - trash_generation * TIME_THRASH_GENERATION),
time_landfill)
history.extend([i | {'day': DaysOfTheWeek(day % 7)} for i in temp])
global_time = trash_generation * TIME_THRASH_GENERATION
else:
global_time = global_time + (total_time_spend // NUMBER_OF_TRUCKS)
# [ 0, 8, 16, ]
# TRASH generation protocol
if sum(trash_in_block) == 0 or global_time >= trash_generation * TIME_THRASH_GENERATION * 60:
trash_generation += 1
if trash_generation == (24 // TIME_THRASH_GENERATION) + 1:
history.append({
"truck id": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"from": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"to": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"time spend to move": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"time left of the truck": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"time spend to collect": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"amount of thrash collected": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"capacity left of the truck": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"thrash left in block": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"thrash left in all blocks": f"finished {day} {DaysOfTheWeek(day % 7).name}",
"total_time_spend": 0})
break
current_day = DaysOfTheWeek(day % 7)
more_thrash_generated = 1 + special_increments.get(current_day, 0)
temp = create_random_list(NUMBER_OF_BLOCKS,
lambda: abs(int(random.gauss(MEAN_THRASH_BLOCK // 3, std))))
total_thrash = sum(temp)
trash_in_block = [(a + b) * more_thrash_generated for a, b in zip(trash_in_block, temp)]
new_total_trash_generated = sum(temp)
history.append({
"truck id": f"More trash generated {new_total_trash_generated} kg",
"from": f"More trash generated {new_total_trash_generated} kg",
"to": f"More trash generated {new_total_trash_generated} kg",
"time spend to move": f"More trash generated {new_total_trash_generated} kg",
"time left of the truck": f"More trash generated {new_total_trash_generated} kg",
"time spend to collect": f"More trash generated {new_total_trash_generated} kg",
"amount of thrash collected": f"More trash generated {new_total_trash_generated} kg",
"capacity left of the truck": f"More trash generated {new_total_trash_generated} kg",
"thrash left in block": f"More trash generated {new_total_trash_generated} kg",
"thrash left in all blocks": f"More trash generated {new_total_trash_generated} kg",
"total_time_spend": 0})
trucks_pools.sort(key=Truck.priority)
for current_truck in trucks_pools:
if current_truck.actual_position != Blocks.LANDFILL:
return_to_landfill(current_truck, time_landfill)
if current_truck.turn % 3 == 0:
current_truck.turn += 1
history.append({
"truck id": f"Day {day + 1} finished {current_truck.event_name}",
"from": f"Day {day + 1} finished {current_truck.event_name}",
"to": f"Day {day + 1} finished {current_truck.event_name}",
"time spend to move": f"Day {day + 1} finished {current_truck.event_name}",
"time left of the truck": f"Day {day + 1} finished {current_truck.event_name}",
"time spend to collect": None,
"amount of thrash collected": None,
"capacity left of the truck": None,
"thrash left in block": None,
"thrash left in all blocks": 0,
"total_time_spend": 0})
with open('history.csv', 'w') as file:
headers = list(history[0].keys())
writer = csv.DictWriter(file, fieldnames=headers)
print(headers)
writer.writeheader()
writer.writerows(history)
it isnot tupla
class Truck:
def __init__(self, name: str,
time_left_hour: int = 8,
capacity_kg: int = 6,
time_coefficient: int = 85,
minimum_time_to_collect: int = 10,
block_sequence: List[Blocks] = None):
"""
:param str name: The truck identifier name
:param int time_left: The time left of the truck
:param int capacity: The capacity left of the truck
:param int time_coefficient: The coefficient of time it takes to collect n kg in 1 minute
:param int minimum_time_to_collect: The minimal time left that a truck need to make i
"""
self.event_name = name
self.time_left = time_left_hour * 60
self.capacity_left = capacity_kg * 1000
self.route = [Blocks(-1)]
self.turn = 1
self.day = 1
self.time_coefficient = time_coefficient # n kg/time
self.minimum_time_to_collect = minimum_time_to_collect
self.actual_position = Blocks(-1)
self.history: List[Dict] = []
what could it be?