Vehicle routing problem change example in Python

[sharmas1ddharth]

I am using timefold and want to add problem change in my code, is there any example that I can use. As, documentation is unclear to me how to use problem change class.

[Christopher-Chianelli]

I agree the docs are a bit unclear, and can probably use an example.
You can look at the examples in our tests:

(Solver)

timefold-solver/python/python-core/tests/test_solver_problem_change.py

Lines 47 to 100 in c42acc7

	class AddEntity(ProblemChange[Solution]):
	entity: Entity
	def __init__(self, entity: Entity):
	self.entity = entity
	def do_change(self, working_solution: Solution, problem_change_director: ProblemChangeDirector):
	problem_change_director.add_entity(self.entity,
	lambda working_entity: working_solution.entities.append(working_entity))
	class RemoveEntity(ProblemChange[Solution]):
	entity: Entity
	def __init__(self, entity: Entity):
	self.entity = entity
	def do_change(self, working_solution: Solution, problem_change_director: ProblemChangeDirector):
	problem_change_director.remove_entity(self.entity,
	lambda working_entity: working_solution.entities.remove(working_entity))
	def test_add_entity():
	solver_config = SolverConfig(
	solution_class=Solution,
	entity_class_list=[Entity],
	score_director_factory_config=ScoreDirectorFactoryConfig(
	constraint_provider_function=maximize_constraints,
	),
	termination_config=TerminationConfig(
	best_score_limit='6'
	)
	)
	problem: Solution = Solution([Entity('A')], [1, 2, 3])
	solver = SolverFactory.create(solver_config).build_solver()
	result: Solution | None = None
	def do_solve(problem: Solution):
	nonlocal solver, result
	result = solver.solve(problem)
	thread = Thread(target=do_solve, args=(problem,), daemon=True)
	thread.start()
	solver.add_problem_change(AddEntity(Entity('B')))
	thread.join(timeout=1)
	if thread.is_alive():
	raise AssertionError(f'Thread {thread} did not finish after 5 seconds')
	assert result is not None
	assert len(result.entities) == 2
	assert result.score.score == 6

(SolverManager)

timefold-solver/python/python-core/tests/test_solver_manager.py

Lines 57 to 100 in c42acc7

	class UseOnlyEntityAndValueProblemChange(ProblemChange[Solution]):
	def __init__(self, entity, value):
	self.entity = entity
	self.value = value
	def do_change(self, solution: Solution, problem_change_director: ProblemChangeDirector):
	problem_facts_to_remove = solution.value_list.copy()
	entities_to_remove = solution.entity_list.copy()
	for problem_fact in problem_facts_to_remove:
	problem_change_director.remove_problem_fact(problem_fact,
	lambda value: solution.value_list.remove(value))
	for removed_entity in entities_to_remove:
	problem_change_director.remove_entity(removed_entity,
	lambda entity: solution.entity_list.remove(entity))
	problem_change_director.add_entity(self.entity, lambda entity: solution.entity_list.append(entity))
	problem_change_director.add_problem_fact(self.value, lambda value: solution.value_list.append(value))
	solver_config = SolverConfig(
	solution_class=Solution,
	entity_class_list=[Entity],
	score_director_factory_config=ScoreDirectorFactoryConfig(
	constraint_provider_function=my_constraints
	),
	termination_config=TerminationConfig(
	best_score_limit='6'
	)
	)
	problem: Solution = Solution([Entity('A'), Entity('B'), Entity('C')], [Value(1), Value(2), Value(3)],
	SimpleScore.ONE)
	def assert_solver_run(solver_manager, solver_job):
	assert solver_manager.get_solver_status(1) != SolverStatus.NOT_SOLVING
	lock.release()
	solution = solver_job.get_final_best_solution()
	assert solution.score.score == 6
	value_list = [entity.value.value for entity in solution.entity_list]
	assert 1 in value_list
	assert 2 in value_list
	assert 3 in value_list
	assert solver_manager.get_solver_status(1) == SolverStatus.NOT_SOLVING
	def assert_problem_change_solver_run(solver_manager, solver_job):
	assert solver_manager.get_solver_status(1) != SolverStatus.NOT_SOLVING
	solver_manager.add_problem_change(1, UseOnlyEntityAndValueProblemChange(Entity('D'), Value(6)))

[Christopher-Chianelli]

To look up locations, the Location class must have an attribute annotated with PlanningId. For example,

class Location:
    id: Annotated[str, PlanningId]
    ...

[sharmas1ddharth]

Now getting this error.

_jpyinterpreter.conversions.TranslatedJavaSystemError: IllegalStateException:  The externalObject ([2688.322600%, 8100.251600%]) with planningId (Pair[key=class org.jpyinterpreter.user.src.domain.location.Location, value=2]) has no known workingObject (null).
Maybe the workingObject was never added because the planning solution doesn't have a @ProblemFactCollectionProperty annotation on a member with instances of the externalObject's class (class org.jpyinterpreter.user.src.domain.location.Location).
java.lang.IllegalStateException: The externalObject ([2688.322600%, 8100.251600%]) with planningId (Pair[key=class org.jpyinterpreter.user.src.domain.location.Location, value=2]) has no known workingObject (null).
Maybe the workingObject was never added because the planning solution doesn't have a @ProblemFactCollectionProperty annotation on a member with instances of the externalObject's class (class org.jpyinterpreter.user.src.domain.location.Location).
	at ai.timefold.solver.core.impl.domain.lookup.PlanningIdLookUpStrategy.lookUpWorkingObject(PlanningIdLookUpStrategy.java:50)
	at ai.timefold.solver.core.impl.domain.lookup.LookUpManager.lookUpWorkingObject(LookUpManager.java:53)
	at ai.timefold.solver.core.impl.score.director.AbstractScoreDirector.lookUpWorkingObject(AbstractScoreDirector.java:540)
	at ai.timefold.solver.core.impl.solver.change.DefaultProblemChangeDirector.lookUpWorkingObjectOrFail(DefaultProblemChangeDirector.java:93)
	at ai.timefold.solver.core.impl.solver.change.DefaultProblemChangeDirector.changeProblemProperty(DefaultProblemChangeDirector.java:84)
	at org.jpype.proxy.JPypeProxy.hostInvoke(Native Method)
	at org.jpype.proxy.JPypeProxy.invoke(Unknown Source)
	at jdk.proxy2/jdk.proxy2.$Proxy33.doChange(Unknown Source)
	at ai.timefold.solver.core.impl.solver.change.ProblemChangeAdapter.lambda$create$1(ProblemChangeAdapter.java:21)
	at ai.timefold.solver.core.impl.solver.DefaultSolver.checkProblemFactChanges(DefaultSolver.java:339)
	at ai.timefold.solver.core.impl.solver.DefaultSolver.solve(DefaultSolver.java:207)
	at ai.timefold.solver.core.impl.solver.DefaultSolverJob.call(DefaultSolverJob.java:125)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)

This is my location class:

class Location(JsonDomainBase):
    id: Annotated[str, PlanningId]
    longitude: Optional[float]
    latitude: Optional[float]
    driving_time_seconds_map: Optional[dict] = {}
    driving_distance_meters_map: Optional[dict] = {}

    def __init__(self, /, **data: Any):
        super().__init__(**data)
        if (
            not hasattr(self, "driving_time_seconds_map")
            or self.driving_time_seconds_map is None
        ):
            self.driving_time_seconds_map = {}
        if (
            not hasattr(self, "driving_distance_meters_map")
            or self.driving_distance_meters_map is None
        ):
            self.driving_distance_meters_map = {}

    def get_latitude(self):
        return self.latitude

    def get_longitude(self):
        return self.longitude

    def get_driving_time_seconds_map(self):
        return self.driving_time_seconds_map

    def driving_time_to(self, other: "Location") -> int:
        return round(self.driving_time_seconds_map[(other.longitude, other.latitude)])

    def driving_distance_to(self, other: "Location") -> int:
        return round(
            self.driving_distance_meters_map[(other.longitude, other.latitude)]
        )

    def set_driving_time_matrix(self, matrix: Dict["Location", float]):
        for key, val in matrix.items():
            self.driving_time_seconds_map[(key.longitude, key.latitude)] = val

    def set_driving_distance_matrix(self, matrix: Dict["Location", float]):
        for key, val in matrix.items():
            self.driving_distance_meters_map[
                (
                    key.longitude,
                    key.latitude,
                )
            ] = val

[Christopher-Chianelli]

It probably because Location is not in the @planning_solution class; this should work (make sure customers and vehicle has a PlanningId attribute).

 class AddCustomer(ProblemChange[Solution]):
    customer: Customer
    def __init__(self, customer: Customer):
        self.customer = customer

    def do_change(self, working_solution: Solution, problem_change_director: ProblemChangeDirector) -> None:
        problem_change_director.add_entity(self.customer, lambda working_entity: working_solution.customers.append(working_entity))
        for customer in working_solution.customers:
            problem_change_director.change_problem_property(customer, lambda working_customer: update_location_matrix(working_customer.location, self.customer.location))
        for vehicle in working_solution.vehicles:
            problem_change_director.change_problem_property(vehicle, lambda working_vehicle: update_location_matrix(working_vehicle.depot, self.customer.location))

[sharmas1ddharth]

Hey, now there's some other issues arising that 'Location' object has no attribute 'longitude' but it is defined, I have uploaded my code to github can you please see and help me what I am doing wrong and what needs to be changed.

https://github.com/sharmas1ddharth/vrp.git

[sharmas1ddharth]

Hey, can you help me please.