(self, message=None, prefix=False)
| 137 | return spatial_info |
| 138 | |
| 139 | def get_temporal_info(self, message=None, prefix=False)->Dict[str,Any]: |
| 140 | temporal_info = {} |
| 141 | if self.temporal_predecessors is not None: |
| 142 | for predecessor in self.temporal_predecessors: |
| 143 | predecessor_outputs = predecessor.last_memory['outputs'] if message is None else predecessor.last_memory[message]['outputs'] |
| 144 | if isinstance(predecessor_outputs, list) and len(predecessor_outputs): |
| 145 | predecessor_output = predecessor_outputs[-1] |
| 146 | elif isinstance(predecessor_outputs, list) and len(predecessor_outputs)==0 and not prefix: |
| 147 | continue |
| 148 | else: |
| 149 | predecessor_output = predecessor_outputs |
| 150 | temporal_info[predecessor.id] = {"role":predecessor.role,"output":predecessor_output} |
| 151 | |
| 152 | return temporal_info |
| 153 | |
| 154 | def execute(self, input:Any, **kwargs): |
| 155 | """Synchronous execution wrapper that normalizes outputs to a list.""" |
no outgoing calls
no test coverage detected