MCPcopy Index your code
hub / github.com/bhowiebkr/PyFlowGraph / update_pins_from_code

Method update_pins_from_code

src/core/node.py:643–732  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

641 self.remove_pin(current_pins[i])
642
643 def update_pins_from_code(self):
644 new_data_inputs, new_data_outputs = {}, {}
645 self.function_name, main_func_def = None, None
646
647 try:
648 tree = ast.parse(self.code)
649 for node in tree.body:
650 if isinstance(node, ast.FunctionDef):
651 for decorator in node.decorator_list:
652 if isinstance(decorator, ast.Name) and decorator.id == "node_entry":
653 main_func_def = node
654 break
655 if main_func_def:
656 break
657 if not main_func_def:
658 # Remove all pins if no valid function
659 for pin in list(self.pins):
660 self.remove_pin(pin)
661 self.fit_size_to_content()
662 return
663
664 self.function_name = main_func_def.name
665
666 # Parse data input pins from function parameters
667 for arg in main_func_def.args.args:
668 new_data_inputs[arg.arg] = self._parse_type_hint(arg.annotation).lower()
669
670 # Parse data output pins from return annotation
671 if main_func_def.returns:
672 return_annotation = main_func_def.returns
673 if isinstance(return_annotation, ast.Subscript) and isinstance(return_annotation.value, ast.Name) and return_annotation.value.id.lower() == "tuple":
674 # Handle Tuple[str, int, bool] - multiple outputs
675 if hasattr(return_annotation.slice, 'elts'):
676 # Check for named outputs in docstring first
677 named_outputs = self._parse_output_names_from_docstring(main_func_def)
678
679 for i, elt in enumerate(return_annotation.slice.elts):
680 type_name = self._parse_type_hint(elt).lower()
681
682 # Use named output if available, otherwise use generic name
683 if i < len(named_outputs):
684 output_name = named_outputs[i]
685 else:
686 output_name = f"output_{i+1}"
687
688 new_data_outputs[output_name] = type_name
689 else:
690 # Single tuple element like Tuple[str]
691 named_outputs = self._parse_output_names_from_docstring(main_func_def)
692 type_name = self._parse_type_hint(return_annotation.slice).lower()
693
694 if named_outputs:
695 new_data_outputs[named_outputs[0]] = type_name
696 else:
697 new_data_outputs["output_1"] = type_name
698 else:
699 # Handle single return types (including List[Dict], Dict[str, int], etc.)
700 named_outputs = self._parse_output_names_from_docstring(main_func_def)

Callers 5

set_codeMethod · 0.95
executeMethod · 0.80
undoMethod · 0.80
executeMethod · 0.80

Calls 6

remove_pinMethod · 0.95
fit_size_to_contentMethod · 0.95
_parse_type_hintMethod · 0.95
_update_data_pinsMethod · 0.95
add_execution_pinMethod · 0.95