Add pipeline states to ir

forte/data/ontology/code_generation_objects.py

@@ -785,6 +785,54 @@ def collect_parents(self, node_dict: Dict[str, Set[str]]):
  ] = found_node.parent.attributes
  found_node = found_node.parent
 
+ def todict(self) -> Dict[str, Any]:
+ r"""Dump the EntryTree structure to a dictionary.
+
+ Returns:
+ dict: A dictionary storing the EntryTree.
+ """
+
+ def node_to_dict(node: EntryTreeNode):
+ return (
+ None
+ if not node
+ else {
+ "name": node.name,
+ "attributes": list(node.attributes),
+ "children": [
+ node_to_dict(child) for child in node.children
+ ],
+ }
+ )
+
+ return node_to_dict(self.root)
+
+ def fromdict(
+ self, tree_dict: Dict[str, Any], parent_entry_name: Optional[str] = None
+ ) -> Optional["EntryTree"]:
+ r"""Load the EntryTree structure from a dictionary.
+
+ Args:
+ tree_dict: A dictionary storing the EntryTree.
+ parent_entry_name: The type name of the parent of the node to be
+ built. Default value is None.
+ """
+ if not tree_dict:
+ return None
+
+ if parent_entry_name is None:
+ self.root = EntryTreeNode(name=tree_dict["name"])
+ self.root.attributes = set(tree_dict["attributes"])
+ else:
+ self.add_node(
+ curr_entry_name=tree_dict["name"],
+ parent_entry_name=parent_entry_name,
+ curr_entry_attr=set(tree_dict["attributes"]),
+ )
+ for child in tree_dict["children"]:
+ self.fromdict(child, tree_dict["name"])
+ return self
+
 
 def search(node: EntryTreeNode, search_node_name: str):
  if node.name == search_node_name:

forte/data/selector.py

@@ -39,7 +39,7 @@
 
 class Selector(Generic[InputPackType, OutputPackType]):
  def __init__(self, **kwargs):
- pass
+ self._stored_kwargs = kwargs
 
  def select(self, pack: InputPackType) -> Iterator[OutputPackType]:
  raise NotImplementedError
@@ -69,7 +69,7 @@ class NameMatchSelector(SinglePackSelector):
  """
 
  def __init__(self, select_name: str):
- super().__init__()
+ super().__init__(select_name=select_name)
  assert select_name is not None
  self.select_name: str = select_name
 
@@ -90,7 +90,7 @@ class RegexNameMatchSelector(SinglePackSelector):
  r"""Select a :class:`DataPack` from a :class:`MultiPack` using a regex."""
 
  def __init__(self, select_name: str):
- super().__init__()
+ super().__init__(select_name=select_name)
  assert select_name is not None
  self.select_name: str = select_name
 

forte/pipeline.py

@@ -60,6 +60,7 @@
 from forte.processors.base.batch_processor import BaseBatchProcessor
 from forte.utils import create_class_with_kwargs
 from forte.utils.utils_processor import record_types_and_attributes_check
+from forte.version import FORTE_IR_VERSION
 
 if sys.version_info < (3, 7):
  import importlib_resources as resources
@@ -247,16 +248,32 @@ def init_from_config_path(self, config_path):
  configs = yaml.safe_load(open(config_path))
  self.init_from_config(configs)
 
- def init_from_config(self, configs: List):
+ def init_from_config(self, configs: Dict[str, Any]):
  r"""Initialized the pipeline (ontology and processors) from the
  given configurations.
 
  Args:
- configs: The configs used to initialize the pipeline.
+ configs: The configs used to initialize the pipeline. It should be
+ a dictionary that contains `forte_ir_version`, `components`
+ and `states`. `forte_ir_version` is a string used to validate
+ input format. `components` is a list of dictionary that
+ contains `type` (the class of pipeline components),
+ `configs` (the corresponding component's configs) and
+ `selector`. `states` will be used to update the pipeline states
+ based on the fields specified in `states.attribute` and
+ `states.resource`.
  """
+ # Validate IR version
+ if configs.get("forte_ir_version") != FORTE_IR_VERSION:
+ raise ProcessorConfigError(
+ f"forte_ir_version={configs.get('forte_ir_version')} not "
+ "supported. Please make sure the format of input IR complies "
+ f"with forte_ir_version={FORTE_IR_VERSION}."
+ )
 
+ # Add components from IR
  is_first: bool = True
- for component_config in configs:
+ for component_config in configs.get("components", []):
  component = create_class_with_kwargs(
  class_name=component_config["type"],
  class_args=component_config.get("kwargs", {}),
@@ -271,7 +288,32 @@ def init_from_config(self, configs: List):
  is_first = False
  else:
  # Can be processor, caster, or evaluator
- self.add(component, component_config.get("configs", {}))
+ selector_config = component_config.get("selector")
+ self.add(
+ component=component,
+ config=component_config.get("configs", {}),
+ selector=selector_config
+ and create_class_with_kwargs(
+ class_name=selector_config["type"],
+ class_args=selector_config.get("kwargs", {}),
+ ),
+ )
+
+ # Set pipeline states and resources
+ states_config: Dict[str, Dict] = configs.get("states", {})
+ for attr, val in states_config.get("attribute", {}).items():
+ setattr(self, attr, val)
+ resource_config: Dict[str, Dict] = states_config.get("resource", {})
+ if "onto_specs_dict" in resource_config:
+ self.resource.update(
+ onto_specs_dict=resource_config["onto_specs_dict"]
+ )
+ if "merged_entry_tree" in resource_config:
+ self.resource.update(
+ merged_entry_tree=EntryTree().fromdict(
+ resource_config["merged_entry_tree"]
+ ),
+ )
 
  def _dump_to_config(self):
  r"""Serialize the pipeline to an IR(intermediate representation).
@@ -281,24 +323,89 @@ def _dump_to_config(self):
  Returns:
  dict: A dictionary storing IR.
  """
- configs: List[Dict] = []
- configs.append(
+
+ def get_type(instance) -> str:
+ r"""Get full module name of an instance"""
+ return instance.__module__ + "." + type(instance).__name__
+
+ def test_jsonable(test_dict: Dict, type_name: str = ""):
+ r"""Check if a dictionary is JSON serializable"""
+ try:
+ json.dumps(test_dict)
+ return test_dict
+ except (TypeError, OverflowError) as e:
+ raise ProcessorConfigError(
+ f"{type_name} is not JSON serializable. Please double "
+ "check the configuration or arguments"
+ ) from e
+
+ configs: Dict = {
+ "forte_ir_version": FORTE_IR_VERSION,
+ "components": list(),
+ "states": dict(),
+ }
+
+ # Serialize pipeline components
+ configs["components"].append(
  {
- "type": ".".join(
- [self._reader.__module__, type(self._reader).__name__]
+ "type": get_type(self._reader),
+ "configs": test_jsonable(
+ test_dict=self._reader_config.todict(),
+ type_name=f"Configuration of {get_type(self._reader)}",
  ),
- "configs": self._reader_config.todict(),
  }
  )
- for component, config in zip(self.components, self.component_configs):
- configs.append(
+ for component, config, selector in zip(
+ self.components, self.component_configs, self._selectors
+ ):
+ configs["components"].append(
  {
- "type": ".".join(
- [component.__module__, type(component).__name__]
+ "type": get_type(component),
+ "configs": test_jsonable(
+ test_dict=config.todict(),
+ type_name=f"Configuration of {get_type(component)}",
  ),
- "configs": config.todict(),
+ "selector": {
+ "type": get_type(selector),
+ "kwargs": test_jsonable(
+ # pylint: disable=protected-access
+ test_dict=selector._stored_kwargs,
+ # pylint: enable=protected-access
+ type_name=f"kwargs of {get_type(selector)}",
+ ),
+ },
  }
  )
+
+ # Serialize current states of pipeline
+ configs["states"].update(
+ {
+ "attribute": {
+ attr: getattr(self, attr)
+ for attr in (
+ "_initialized",
+ "_enable_profiling",
+ "_check_type_consistency",
+ "_do_init_type_check",
+ )
+ if hasattr(self, attr)
+ },
+ "resource": dict(),
+ }
+ )
+ if self.resource.contains("onto_specs_dict"):
+ configs["states"]["resource"].update(
+ {"onto_specs_dict": self.resource.get("onto_specs_dict")}
+ )
+ if self.resource.contains("merged_entry_tree"):
+ configs["states"]["resource"].update(
+ {
+ "merged_entry_tree": self.resource.get(
+ "merged_entry_tree"
+ ).todict()
+ }
+ )
+
  return configs
 
  def save(self, path: str):

forte/processors/base/base_processor.py

@@ -130,11 +130,6 @@ def default_configs(cls) -> Dict[str, Any]:
  config = super().default_configs()
  config.update(
  {
- "selector": {
- "type": "forte.data.selector.DummySelector",
- "args": None,
- "kwargs": {},
- },
  "overwrite": False,
  }
  )

forte/version.py

@@ -18,3 +18,4 @@
 
 VERSION_SHORT = "{0}.{1}".format(_MAJOR, _MINOR)
 VERSION = "{0}.{1}.{2}".format(_MAJOR, _MINOR, _REVISION)
+FORTE_IR_VERSION = "0.0.1"