MCPcopy Create free account
hub / github.com/caelan/pddlstream / FunctionInstance

Class FunctionInstance

pddlstream/language/function.py:66–120  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

64 return '{}={:.3f}'.format(str_from_head(self.instance.head), self.value)
65
66class FunctionInstance(Instance):
67 _Result = FunctionResult
68 #_opt_value = 0
69 def __init__(self, external, input_objects):
70 super(FunctionInstance, self).__init__(external, input_objects)
71 self._head = None
72 @property
73 def head(self):
74 if self._head is None:
75 self._head = substitute_expression(self.external.head, self.mapping)
76 return self._head
77 @property
78 def value(self):
79 assert len(self.history) == 1
80 return self.history[0]
81 def _compute_output(self):
82 self.enumerated = True
83 self.num_calls += 1
84 if self.history:
85 return self.value
86 input_values = self.get_input_values()
87 value = self.external.fn(*input_values)
88 # TODO: cast the inputs and test whether still equal?
89 # if not (type(self.value) is self.external._codomain):
90 # if not isinstance(self.value, self.external.codomain):
91 if value < 0:
92 raise ValueError('Function [{}] produced a negative value [{}]'.format(self.external.name, value))
93 self.history.append(self.external.codomain(value))
94 return self.value
95 def next_results(self, verbose=False):
96 assert not self.enumerated
97 start_time = time.time()
98 start_history = len(self.history)
99 value = self._compute_output()
100 new_results = [self._Result(self, value, optimistic=False)]
101 new_facts = []
102
103 if (value is not False) and verbose:
104 # TODO: str(new_results[-1])
105 print('iter={}, outs={}) {}{}={:.3f}'.format(
106 self.get_iteration(), len(new_results), get_prefix(self.external.head),
107 str_from_object(self.get_input_values()), value))
108 if start_history <= len(self.history) - 1:
109 self.update_statistics(start_time, new_results)
110 self.successful |= any(r.is_successful() for r in new_results)
111 return new_results, new_facts
112 def next_optimistic(self):
113 if self.enumerated or self.disabled:
114 return []
115 # TODO: cache this value
116 opt_value = self.external.opt_fn(*self.get_input_values())
117 self.opt_results = [self._Result(self, opt_value, optimistic=True)]
118 return self.opt_results
119 def __repr__(self):
120 return '{}=?{}'.format(str_from_head(self.head), self.external.codomain.__name__)
121
122class Function(External):
123 """

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected