MA-PDDL writer #266
MA-PDDL writer #266
Conversation
Codecov ReportBase: 90.40% // Head: 87.88% // Decreases project coverage by
Additional details and impacted files@@ Coverage Diff @@
## master #266 +/- ##
==========================================
- Coverage 90.40% 87.88% -2.52%
==========================================
Files 133 136 +3
Lines 16314 16866 +552
==========================================
+ Hits 14748 14822 +74
- Misses 1566 2044 +478
Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here. ☔ View full report at Codecov. |
There was a problem hiding this comment.
@Alee08 @fabio-patrizi Unfortunately, I think this requires a bit more work: see my comments for the details but from the high-level POV you need to avoid creating needless expressions just for printing (and I think there are bugs in the current creation of such expressions) and fix the interface with streams.
Moreover, as a general point, you need to rebase the code to the current master and check the use of get_mangled_name that is not always used on the agents names.
| def convert_to_dot_exp( | ||
| self, node: "up.model.FNode", agent: "up.model.multi_agent.Agent" | ||
| ): | ||
| """ | ||
| This function simplifies an FNode to convert it into an FNode of type Dot and substitute the new FNode for the original expression. | ||
| """ | ||
| # Initialize a simplifier object | ||
| o = Simplifier(self.problem.env) | ||
| # Define a helper function to simplify argument | ||
| def simplify_arg(arg): | ||
| # Simplify the argument | ||
| arg = o.simplify(arg) | ||
| if arg.is_not(): | ||
| arg = self.problem.env.expression_manager.Not(arg) | ||
| return arg | ||
|
|
||
| # Initialize an empty list to store simplified arguments | ||
| args_list = [] | ||
| # Iterate through all arguments of the node | ||
| for arg in node.args: | ||
| if node.is_fluent_exp(): | ||
| # If the fluent is in the agent's fluents | ||
| if node.fluent() in agent.fluents: | ||
| # Create a new dot expression | ||
| new_dot_expression = self.problem.env.expression_manager.Dot( | ||
| agent, node | ||
| ) | ||
| return new_dot_expression | ||
| elif len(arg.args) > 1: | ||
| # Extend args_list with both sub-arguments simplified | ||
| args_list.extend([simplify_arg(arg.args[0]), simplify_arg(arg.args[1])]) | ||
| else: | ||
| # Append arg to args_list after simplify | ||
| args_list.append(simplify_arg(arg)) | ||
|
|
||
| # Create a new dot expression from the list of simplified args | ||
| new_dot_expression = self.sub_exp_dot(args_list, agent, node) | ||
| return new_dot_expression | ||
|
|
||
| def sub_exp( | ||
| self, | ||
| old_exp: "up.model.FNode", | ||
| new_exp: "up.model.FNode", | ||
| expression: "up.model.FNode", | ||
| ): | ||
| subs_map: Dict[Expression, Expression] = {} | ||
| sub = Substituter(self.problem.env) | ||
| subs_map[old_exp] = new_exp | ||
| new_expression = sub.substitute(expression, subs_map) | ||
| return new_expression | ||
|
|
||
| def sub_exp_dot( | ||
| self, | ||
| list_old_exp: List["up.model.FNode"], | ||
| agent: "up.model.multi_agent.Agent", | ||
| expression, | ||
| ): | ||
| subs_map: Dict[Expression, Expression] = {} | ||
| sub = Substituter(self.problem.env) | ||
| for old_exp in list_old_exp: | ||
| if old_exp.fluent() in agent.fluents: | ||
| arg_dot = self.problem.env.expression_manager.Dot(agent, old_exp) | ||
| subs_map[old_exp] = arg_dot | ||
| else: | ||
| return old_exp | ||
| new_expression = sub.substitute(expression, subs_map) | ||
| return new_expression | ||
|
|
||
| def add_agent_prefix(self, string: str, fluents_list: List["Fluent"]): | ||
| match = re.search(r"\((\w+)\s\?", string) | ||
| if match: | ||
| found = match.group(1) | ||
| else: | ||
| return string | ||
| for obj in fluents_list: | ||
| if obj.name == found: | ||
| return re.sub(r"\((\w+)\s\?", r"(a_\1 ?", string) | ||
| return string |
There was a problem hiding this comment.
- These methods are very strange for a printer: why do you need to build new expressions? I think it would be MUCH simpler to add an optional
agentparameter to theConverterToMAPDDLStringand add the agent name where needed. For example:
class ConverterToMAPDDLString(ConverterToPDDLString):
"""Expression converter to a MA-PDDL string."""
def __init__(
self,
env: "up.environment.Environment",
get_mangled_name: Callable[
[
Union[
"up.model.Type",
"up.model.Action",
"up.model.Fluent",
"up.model.Object",
"up.model.multi_agent.Agent",
]
],
str,
],
agent : Optional["up.model.multi_agent.Agent"]
):
ConverterToPDDLString.__init__(self, env, get_mangled_name)
self.agent = agent
def walk_dot(self, expression, args):
agent = expression.agent()
fluent = expression.args[0].fluent()
objects = expression.args[0].args
return f'({self.get_mangled_name(fluent)} {self.get_mangled_name(agent)} {" ".join([self.convert(obj) for obj in objects])})'
def walk_fluent_exp(self, expression, args):
fluent = expression.fluent()
return f'({self.get_mangled_name(fluent)} {self.get_mangled_name(self.agent) if self.agent is not None and fluent in self.agent.fluents else ""}{" " if len(args) > 0 else ""}{" ".join(args)})'In this way, you can simply create different ConverterToMAPDDLString objects for different agents and use them. This should greatly simplify the printing, because you simply call the appropriate printer on a problem expression, without creating new FNodes.
- these functions seems seriously buggy: why adding a
Notif an expression is aNot? What happens if an expression has more than two arguments?sub_expis never used....
There was a problem hiding this comment.
The simplify() method aims to simplify the expression but does not take into account the "Not" operator. This is why I check if the FNode is of type "Not" and if it is, I remove it using self.problem.env.expression_manager.Not(arg). This function will remove the "Not" operator if it is present, otherwise, it adds it.
I agree with your suggestion to use ConverterToMAPDDLString as it simplifies the printing process.
Done!
| self.domain_objects: Optional[Dict[_UserType, Set[Object]]] = None | ||
| self.domain_objects_agents: Dict[up.model.multi_agent.Agent, str] | ||
| self.domain_fluents_agents: Dict[up.model.multi_agent.Agent, List[Fluent]] |
There was a problem hiding this comment.
What is the purpose of this state? The Printer class should ideally be reusable across multiple problems and adding state requires cleanup (that seems to be missing)
There was a problem hiding this comment.
The states domain_objects_agents and domain_fluents_agents are used to keep track of fluents and objects of other agents that are referenced using the DOT operator in the preconditions. They allow the addition of these fluents and objects to the public predicates and constants in the MA-PDDL domain of the current agent. This is necessary since these fluents are not present in the current agent, and the current agent's fluents do not use the objects.
| f"The item {item} does not correspond to any item renamed." | ||
| ) | ||
|
|
||
| def _populate_domain_objects( |
There was a problem hiding this comment.
What is the purpose of this function? Please an an explanation comment
There was a problem hiding this comment.
The purpose of this function is to populate the domain_objects, domain_objects_agents, and domain_fluents_agents class variables. These variables are used to keep track of the objects and fluents of other agents that are referenced using the DOT operator in the preconditions of the actions of the current agent. The function iterates through the actions of the current agent and checks the preconditions for any references to other agents and fluents (DOT).
If found, it updates the class variables with the relevant information. This information is later used to add the fluents and objects to the public predicates and constants in the MA-PDDL domain of the current agent. This function is also present in pddl_writer.py and allows you to update the domain_objects (objects present in the preconditions of the actions).
| out.seek(0) | ||
| ag_problem = out.read() | ||
| ag_problems[ag.name] = ag_problem | ||
| out.truncate(0) | ||
| out.seek(0) |
There was a problem hiding this comment.
WTF?
Why do you take out in input at all? You can simply use a StringIO for every agent and return their .getvalue()...
There was a problem hiding this comment.
BTW, you are not guaranteed to have seek and truncate for every type of IO[str], only for files and buffers...
There was a problem hiding this comment.
Thanks for the feedback, I understand that using one StringIO for each agent and returning their .getvalue() is much easier. Initially I wanted the multi-agent _write_domain() to take the output as a parameter and thus look similar to the single-agent one.
Done!
| out.seek(0) | ||
| ag_domain = out.read() | ||
| ag_domains[ag.name] = ag_domain | ||
| out.truncate(0) | ||
| out.seek(0) |
There was a problem hiding this comment.
See comment below: this is very awkward...
|
@Alee08 It seems that you had some problems with the rebase and several unrelated commits entered the branch. |
|
@Alee08 It seems that there are still several unrelated commits (747 commits in total) in this PR. Let us know if you want that we will give a try to fix that. |
@alvalentini @mikand yes thanks. I created a new PR MulitAgent-io #324 equal to this MA-PDDL writer #266, if the current branch is more difficult to merge. |
No description provided.