Add new validate in python (#4982)

Author: njooma

cli/module_generate/_templates/python/src/models/tmpl-resource.py

@@ -1,4 +1,4 @@
-from typing import ClassVar, Mapping, Sequence, Self
+from typing import ClassVar, Mapping, Sequence, Self, Tuple
 from viam.proto.app.robot import ComponentConfig
 from viam.proto.common import ResourceName
 from viam.resource.base import ResourceBase
@@ -25,17 +25,19 @@ def new(cls, config: ComponentConfig, dependencies: Mapping[ResourceName, Resour
         return super().new(config, dependencies)
 
     @classmethod
-    def validate_config(cls, config: ComponentConfig) -> Sequence[str]:
+    def validate_config(cls, config: ComponentConfig) -> Tuple[Sequence[str], Sequence[str]]:
         """This method allows you to validate the configuration object received from the machine,
-        as well as to return any implicit dependencies based on that `config`.
+        as well as to return any required dependencies or optional dependencies based on that `config`.
 
         Args:
             config (ComponentConfig): The configuration for this resource
 
         Returns:
-            Sequence[str]: A list of implicit dependencies
+            Tuple[Sequence[str], Sequence[str]]: A tuple where the 
+                first element is a list of required dependencies and the
+                second element is a list of optional dependencies
         """
-        return []
+        return [], []
 
     def reconfigure(self, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]):
         """This method allows you to dynamically update your service when it receives a new `config` object.

cli/module_generate/scripts/generate_stubs.py

@@ -10,14 +10,15 @@ def return_attribute(value: str, attr: str) -> ast.Attribute:
     return ast.Attribute(
         value=ast.Name(id=value, ctx=ast.Load()),
         attr=attr,
-        ctx=ast.Load())
+        ctx=ast.Load(),
+    )
 
 
 def update_annotation(
     resource_name: str,
     annotation: Union[ast.Name, ast.Subscript],
     nodes: Set[str],
-    parent: str
+    parent: str,
 ) -> Union[ast.Attribute, ast.Subscript]:
     if isinstance(annotation, ast.Name) and annotation.id in nodes:
         value = parent if parent else resource_name
@@ -27,48 +28,43 @@ def update_annotation(
             resource_name,
             annotation.slice,
             nodes,
-            parent)
+            parent,
+        )
     return annotation
 
 
 def replace_async_func(
-    resource_name: str,
-    func: ast.AsyncFunctionDef,
-    nodes: Set[str],
-    parent: str = ""
+    resource_name: str, func: ast.AsyncFunctionDef, nodes: Set[str], parent: str = ""
 ) -> None:
     for arg in func.args.args:
-        arg.annotation = update_annotation(
-            resource_name,
-            arg.annotation,
-            nodes,
-            parent)
+        arg.annotation = update_annotation(resource_name, arg.annotation, nodes, parent)
     func.body = [
-        ast.Expr(ast.Call(
-            func=ast.Attribute(
-                value=ast.Attribute(
-                    ast.Name(id="self"),
-                    attr="logger"),
-                attr="error",
-            ),
-            args=[ast.Constant(value=f"`{func.name}` is not implemented")],
-            keywords=[])),
+        ast.Expr(
+            ast.Call(
+                func=ast.Attribute(
+                    value=ast.Attribute(ast.Name(id="self"), attr="logger"),
+                    attr="error",
+                ),
+                args=[ast.Constant(value=f"`{func.name}` is not implemented")],
+                keywords=[],
+            )
+        ),
         ast.Raise(
-            exc=ast.Call(func=ast.Name(id='NotImplementedError',
-                                       ctx=ast.Load()),
-                         args=[],
-                         keywords=[]),
-            cause=None)
+            exc=ast.Call(
+                func=ast.Name(id="NotImplementedError", ctx=ast.Load()),
+                args=[],
+                keywords=[],
+            ),
+            cause=None,
+        ),
     ]
     func.decorator_list = []
     if isinstance(func.returns, (ast.Name, ast.Subscript)):
-        func.returns = update_annotation(
-            resource_name, func.returns, nodes, parent
-        )
+        func.returns = update_annotation(resource_name, func.returns, nodes, parent)
 
 
 def return_subclass(
-        resource_name: str, stmt: ast.ClassDef, parent: str = ""
+    resource_name: str, stmt: ast.ClassDef, parent: str = ""
 ) -> List[str]:
     def parse_subclass(resource_name: str, stmt: ast.ClassDef, parent: str):
         nodes = set()
@@ -93,9 +89,7 @@ def parse_subclass(resource_name: str, stmt: ast.ClassDef, parent: str):
             stmt.body = [ast.Pass()]
 
     parse_subclass(resource_name, stmt, parent)
-    return '\n'.join(
-        ['    ' + line for line in ast.unparse(stmt).splitlines()]
-    )
+    return "\n".join(["    " + line for line in ast.unparse(stmt).splitlines()])
 
 
 def main(
@@ -108,14 +102,12 @@ def main(
     import isort
     from slugify import slugify
 
-    module_name = (
-        f"viam.{resource_type}s.{resource_subtype}.{resource_subtype}"
-    )
+    module_name = f"viam.{resource_type}s.{resource_subtype}.{resource_subtype}"
     module = import_module(module_name)
-    resource_name = {
-        "input": "Controller", "slam": "SLAM", "mlmodel": "MLModel"
-    }.get(resource_subtype, "".join(word.capitalize()
-                                    for word in resource_subtype.split("_")))
+    resource_name = {"input": "Controller", "slam": "SLAM", "mlmodel": "MLModel"}.get(
+        resource_subtype,
+        "".join(word.capitalize() for word in resource_subtype.split("_")),
+    )
 
     imports, subclasses, abstract_methods = [], [], []
     nodes = set()
@@ -132,8 +124,11 @@ def main(
                 for imp in stmt.names:
                     if imp.name in modules_to_ignore:
                         continue
-                    imports.append(f"import {imp.name} as {imp.asname}"
-                                   if imp.asname else f"import {imp.name}")
+                    imports.append(
+                        f"import {imp.name} as {imp.asname}"
+                        if imp.asname
+                        else f"import {imp.name}"
+                    )
             elif (
                 isinstance(stmt, ast.ImportFrom)
                 and stmt.module
@@ -159,8 +154,8 @@ def main(
                         nodes.add(cstmt.target.id)
                     elif isinstance(cstmt, ast.AsyncFunctionDef):
                         replace_async_func(resource_name, cstmt, nodes)
-                        indented_code = '\n'.join(
-                            ['    ' + line for line in ast.unparse(cstmt).splitlines()]
+                        indented_code = "\n".join(
+                            ["    " + line for line in ast.unparse(cstmt).splitlines()]
                         )
                         abstract_methods.append(indented_code)
 
@@ -172,22 +167,24 @@ def main(
                 for cstmt in stmt.body:
                     if isinstance(cstmt, ast.AsyncFunctionDef):
                         replace_async_func("", cstmt, [])
-                        indented_code = '\n'.join(
-                            ['    ' + line for line in ast.unparse(cstmt).splitlines()]
+                        indented_code = "\n".join(
+                            ["    " + line for line in ast.unparse(cstmt).splitlines()]
                         )
                         abstract_methods.append(indented_code)
                         if cstmt.name == "do_command":
                             imports.append("from typing import Optional")
                             imports.append("from viam.utils import ValueTypes")
                         elif cstmt.name == "get_geometries":
-                            imports.append("from typing import Any, Dict, List, Optional")
+                            imports.append(
+                                "from typing import Any, Dict, List, Optional"
+                            )
                             imports.append("from viam.proto.common import Geometry")
 
     model_name_pascal = "".join(
         [word.capitalize() for word in slugify(model_name).split("-")]
     )
     resource_file = '''
-from typing import ClassVar, Mapping, Sequence
+from typing import ClassVar, Mapping, Sequence, Tuple
 from typing_extensions import Self
 from viam.proto.app.robot import ComponentConfig
 from viam.proto.common import ResourceName
@@ -218,17 +215,19 @@ def new(cls, config: ComponentConfig, dependencies: Mapping[ResourceName, Resour
         return super().new(config, dependencies)
 
     @classmethod
-    def validate_config(cls, config: ComponentConfig) -> Sequence[str]:
+    def validate_config(cls, config: ComponentConfig) -> Tuple[Sequence[str], Sequence[str]]:
         """This method allows you to validate the configuration object received from the machine,
-        as well as to return any implicit dependencies based on that `config`.
+        as well as to return any required dependencies or optional dependencies based on that `config`.
 
         Args:
             config (ComponentConfig): The configuration for this resource
 
         Returns:
-            Sequence[str]: A list of implicit dependencies
+            Tuple[Sequence[str], Sequence[str]]: A tuple where the 
+                first element is a list of required dependencies and the
+                second element is a list of optional dependencies
         """
-        return []
+        return [], []
 
     def reconfigure(self, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]):
         """This method allows you to dynamically update your service when it receives a new `config` object.
@@ -250,8 +249,8 @@ def reconfigure(self, config: ComponentConfig, dependencies: Mapping[ResourceNam
         namespace,
         mod_name,
         model_name,
-        '\n\n'.join([subclass for subclass in subclasses]),
-        '\n\n'.join([f'{method}' for method in abstract_methods]),
+        "\n\n".join([subclass for subclass in subclasses]),
+        "\n\n".join([f"{method}" for method in abstract_methods]),
     )
     f_name = os.path.join(mod_name, "src", "models", "resource.py")
     with open(f_name, "w+") as f:
@@ -274,12 +273,7 @@ def reconfigure(self, config: ComponentConfig, dependencies: Mapping[ResourceNam
     if sys.argv[2] == "mlmodel":
         packages.append("numpy")
     install_res = subprocess.run(
-        [
-            sys.executable,
-            "-m",
-            "pip",
-            "install"
-        ] + packages,
+        [sys.executable, "-m", "pip", "install"] + packages,
         capture_output=True,
     )
     if install_res.returncode != 0: