Method post_validate

src/typings/config.py:69–160

(cls, instance: "AssignmentConfig")

    @classmethod
    def post_validate(cls, instance: "AssignmentConfig"):

        REMOVE_UNUSED_IN_DEFINITION = True
        REMOVE_UNUSED_IN_CONCURRENCY = True

        # Step 1. Check if all agents and tasks are defined, and remove unused agents and tasks

        agent_in_assignment = set()
        task_in_assignment = set()
        for assignment in instance.assignments:
            assert (
                assignment.agent in instance.definition.agent
            ), f"Agent {assignment.agent} is not defined."
            agent_in_assignment.add(assignment.agent)
            assert (
                assignment.task in instance.definition.task
            ), f"Task {assignment.task} is not defined."
            task_in_assignment.add(assignment.task)

        for agent in agent_in_assignment:
            assert (
                agent in instance.concurrency.agent
            ), f"Concurrency of {agent} is not specified."
        for task in task_in_assignment:
            assert (
                task in instance.concurrency.task
            ), f"Concurrency of {task} is not specified."

        def remove_unused(
            target: Union[DefinitionConfig, ConcurrencyConfig], warning_suffix: str
        ):
            nonlocal agent_in_assignment, task_in_assignment
            removed_agents = set()
            removed_tasks = set()

            for definition_agent in target.agent.keys():
                if definition_agent not in agent_in_assignment:
                    removed_agents.add(definition_agent)

            for definition_task in target.task.keys():
                if definition_task not in task_in_assignment:
                    removed_tasks.add(definition_task)

            if len(removed_agents) > 0 or len(removed_tasks) > 0:

                print(
                    ColorMessage.yellow(
                        f"Warning: {len(removed_agents)} agent(s) and {len(removed_tasks)} task(s) are "
                        + warning_suffix
                    ),
                    file=sys.stderr,
                )
                print(ColorMessage.yellow(f"    Agent: {removed_agents}"))
                print(ColorMessage.yellow(f"    Task: {removed_tasks}"))

                for agent in removed_agents:
                    target.agent.pop(agent)