LRPC-9 Use PyLint

* Fixing more MyPy and PyLint issues
* Added logging to report definition warnings and errors
* Normalized loading LrpcDef from various sources

package/lrpc/client/__init__.py

@@ -1,5 +1,5 @@
-from .encoder import lrpc_encode
-from .decoder import lrpc_decode
-from .decoder import LrpcDecoder
-from .lrpc_client import LrpcClient
-from .client_cli_visitor import ClientCliVisitor
+from .encoder import lrpc_encode as lrpc_encode
+from .decoder import lrpc_decode as lrpc_decode
+from .decoder import LrpcDecoder as LrpcDecoder
+from .lrpc_client import LrpcClient as LrpcClient
+from .client_cli_visitor import ClientCliVisitor as ClientCliVisitor

package/lrpc/client/client_cli_visitor.py

@@ -18,16 +18,20 @@ class YamlParamType(click.ParamType):
 
     name = "YAML"
 
+    # function does not always return
+    # pylint: disable=inconsistent-return-statements
     def convert(self, value: Any, param: Optional[click.Parameter], ctx: Optional[click.Context]) -> dict[str, Any]:
         if isinstance(value, dict):
             return value
 
         try:
-            return yaml.safe_load(value)
+            v = yaml.safe_load(value)
+            if isinstance(v, dict):
+                return v
         except yaml.parser.ParserError:
-            self.fail(f"{value} does not contain valid YAML", param, ctx)
+            pass
 
-        return {}
+        self.fail(f"{value} does not contain valid YAML", param, ctx)
 
 
 class OptionalParamType(click.ParamType):

package/lrpc/client/decoder.py

@@ -101,6 +101,7 @@ def __unpack(self, pack_format: str) -> Any:
         self.start += struct.calcsize(pack_format)
         return unpacked[0]
 
+    # pylint: disable = too-many-return-statements
     def lrpc_decode(self, var: LrpcVar) -> Any:
         if var.is_array_of_strings():
             return self.__decode_array_of_strings(var)

package/lrpc/client/lrpc_client.py

@@ -1,14 +1,14 @@
 import struct
 from typing import Any
 
+from ..core.definition import LrpcDef
 from .decoder import LrpcDecoder
 from .encoder import lrpc_encode
-from ..utils import load_lrpc_def
 
 
 class LrpcClient:
-    def __init__(self, definition_file: str) -> None:
-        self.lrpc_def = load_lrpc_def(definition_file)
+    def __init__(self, lrpc_def: LrpcDef) -> None:
+        self.lrpc_def = lrpc_def
         self.receive_buffer = b""
 
     def process(self, encoded: bytes) -> dict[str, Any]:

package/lrpc/codegen/__init__.py

@@ -1,6 +1,6 @@
-from .constants import ConstantsFileVisitor
-from .enum import EnumFileVisitor
-from .server_include import ServerIncludeVisitor
-from .service_include import ServiceIncludeVisitor
-from .service_shim import ServiceShimVisitor
-from .struct import StructFileVisitor
+from .constants import ConstantsFileVisitor as ConstantsFileVisitor
+from .enum import EnumFileVisitor as EnumFileVisitor
+from .server_include import ServerIncludeVisitor as ServerIncludeVisitor
+from .service_include import ServiceIncludeVisitor as ServiceIncludeVisitor
+from .service_shim import ServiceShimVisitor as ServiceShimVisitor
+from .struct import StructFileVisitor as StructFileVisitor

package/lrpc/codegen/service_shim.py

@@ -7,6 +7,7 @@
 from ..core import LrpcDef, LrpcService, LrpcFun, LrpcVar
 
 
+# pylint: disable = too-many-instance-attributes
 class ServiceShimVisitor(LrpcVisitor):
 
     def __init__(self, output: os.PathLike[str]) -> None:

package/lrpc/core/__init__.py

@@ -1,7 +1,7 @@
-from .var import LrpcVar, LrpcVarDict
-from .enum import LrpcEnum, LrpcEnumField, LrpcEnumDict
-from .struct import LrpcStruct, LrpcStructDict
-from .constant import LrpcConstant, LrpcConstantDict
-from .function import LrpcFun, LrpcFunDict
-from .service import LrpcService, LrpcServiceDict
-from .definition import LrpcDef
+from .var import LrpcVar as LrpcVar, LrpcVarDict as LrpcVarDict
+from .enum import LrpcEnum as LrpcEnum, LrpcEnumField as LrpcEnumField, LrpcEnumDict as LrpcEnumDict
+from .struct import LrpcStruct as LrpcStruct, LrpcStructDict as LrpcStructDict
+from .constant import LrpcConstant as LrpcConstant, LrpcConstantDict as LrpcConstantDict
+from .function import LrpcFun as LrpcFun, LrpcFunDict as LrpcFunDict
+from .service import LrpcService as LrpcService, LrpcServiceDict as LrpcServiceDict
+from .definition import LrpcDef as LrpcDef, LrpcDefDict as LrpcDefDict

package/lrpc/core/definition.py

@@ -20,6 +20,7 @@ class LrpcDefDict(TypedDict):
     constants: NotRequired[list[LrpcConstantDict]]
 
 
+# pylint: disable = too-many-instance-attributes
 class LrpcDef:
 
     def __init__(self, raw: LrpcDefDict) -> None:
@@ -75,7 +76,10 @@ def __init_service_ids(raw: LrpcDefDict) -> None:
             if "id" in s:
                 last_service_id = s["id"]
             else:
-                last_service_id = last_service_id + 1
+                # id field must be present in LrpcServiceDict to construct LrpcService
+                # but it is optional when constructing LrpcDef. This method
+                # makes sure that service IDs are initialized properly
+                last_service_id = last_service_id + 1  # type: ignore[unreachable]
                 s["id"] = last_service_id
 
     def accept(self, visitor: LrpcVisitor) -> None:

package/lrpc/core/enum.py

@@ -4,6 +4,11 @@
 from ..visitors import LrpcVisitor
 
 
+class LrpcEnumFieldSimpleDict(TypedDict):
+    name: str
+    id: NotRequired[int]
+
+
 class LrpcEnumFieldDict(TypedDict):
     name: str
     id: int
@@ -17,10 +22,6 @@ def __init__(self, raw: LrpcEnumFieldDict) -> None:
         self.__name = raw["name"]
         self.__id = raw["id"]
 
-    @classmethod
-    def from_name_and_index(cls, name: str, index: int) -> "LrpcEnumField":
-        return cls({"name": name, "id": index})
-
     def name(self) -> str:
         return self.__name
 
@@ -30,7 +31,7 @@ def id(self) -> int:
 
 class LrpcEnumDict(TypedDict):
     name: str
-    fields: list[Union[LrpcEnumFieldDict, str]]
+    fields: list[Union[LrpcEnumFieldSimpleDict, str]]
     external: NotRequired[str]
     external_namespace: NotRequired[str]
 
@@ -62,11 +63,16 @@ def fields(self) -> list[LrpcEnumField]:
         index = 0
         for field in self.__fields:
             if isinstance(field, str):
-                all_fields.append(LrpcEnumField.from_name_and_index(field, index))
+                f = field
+                i = index
             elif "id" not in field:
-                all_fields.append(LrpcEnumField.from_name_and_index(field["name"], index))
+                f = field["name"]
+                i = index
             else:
-                all_fields.append(LrpcEnumField(field))
+                f = field["name"]
+                i = field["id"]
+
+            all_fields.append(LrpcEnumField({"name": f, "id": i}))
 
             index = all_fields[-1].id() + 1
 

package/lrpc/core/service.py

@@ -30,7 +30,10 @@ def __init_functions_ids(functions: list[LrpcFunDict]) -> None:
             if "id" in f:
                 last_function_id = f["id"]
             else:
-                last_function_id = last_function_id + 1
+                # id field must be present in LrpcFunDict to construct LrpcFun
+                # but it is optional when constructing LrpcService. This method
+                # makes sure that function IDs are initialized properly
+                last_function_id = last_function_id + 1  # type: ignore[unreachable]
                 f["id"] = last_function_id
 
     def accept(self, visitor: LrpcVisitor) -> None:

package/lrpc/core/var.py

@@ -24,6 +24,7 @@ class LrpcVarDict(TypedDict):
     count: NotRequired[Union[int, Literal["?"]]]
 
 
+# pylint: disable = too-many-public-methods
 class LrpcVar:
     def __init__(self, raw: LrpcVarDict) -> None:
         assert "name" in raw and isinstance(raw["name"], str)
@@ -217,6 +218,7 @@ def pack_type(self) -> str:
 
     def contained(self) -> "LrpcVar":
         contained_item = deepcopy(self)
-        contained_item.__is_optional = False  # pylint: disable=protected-access, unused-private-member
-        contained_item.__count = 1  # pylint: disable=protected-access, unused-private-member
+        # pylint: disable=protected-access, unused-private-member
+        contained_item.__is_optional = False
+        contained_item.__count = 1
         return contained_item

package/lrpc/lotusrpc.py

@@ -1,13 +1,9 @@
+import logging
 import os
 from os import path
-from typing import TextIO, Any
-
+from typing import TextIO
 import click
-import jsonschema
-import jsonschema.exceptions
-import yaml
 from .visitors import PlantUmlVisitor
-from .schema import load_lrpc_schema
 from .codegen import (
     ConstantsFileVisitor,
     EnumFileVisitor,
@@ -17,44 +13,16 @@
     StructFileVisitor,
 )
 from .core import LrpcDef
-from .validation import SemanticAnalyzer
+from .utils import load_lrpc_def_from_file
+
+logging.basicConfig(level=logging.INFO)
 
 
 def create_dir_if_not_exists(target_dir: os.PathLike[str]) -> None:
     if not path.exists(target_dir):
         os.makedirs(target_dir, 511, True)
 
 
-def validate_yaml(definition: dict[str, Any], input_file_name: str) -> bool:
-    try:
-        jsonschema.validate(definition, load_lrpc_schema())
-        return False
-    except jsonschema.exceptions.ValidationError as e:
-        print("#" * 80)
-        print(" LRPC definition parsing error ".center(80, "#"))
-        print(f" {input_file_name} ".center(80, "#"))
-        print("#" * 80)
-        print(e)
-        return True
-
-
-def validate_definition(lrpc_def: LrpcDef, warnings_as_errors: bool) -> bool:
-    errors_found = False
-
-    sa = SemanticAnalyzer(lrpc_def)
-    sa.analyze()
-
-    if warnings_as_errors:
-        for e in sa.errors + sa.warnings:
-            errors_found = True
-            print(f"Error: {e}")
-    else:
-        for w in sa.warnings:
-            print(f"Warning: {w}")
-
-    return errors_found
-
-
 def generate_rpc(lrpc_def: LrpcDef, output: os.PathLike[str]) -> None:
     create_dir_if_not_exists(output)
 
@@ -76,16 +44,16 @@ def generate_rpc(lrpc_def: LrpcDef, output: os.PathLike[str]) -> None:
 def generate(warnings_as_errors: bool, output: os.PathLike[str], input_file: TextIO) -> None:
     """Generate code for file INPUT"""
 
-    definition = yaml.safe_load(input_file)
-    errors_found = validate_yaml(definition, input_file.name)
-    if errors_found:
-        return
-
-    lrpc_def = LrpcDef(definition)
-    errors_found = validate_definition(lrpc_def, warnings_as_errors)
-    if not errors_found:
-        input_file.seek(0)
+    try:
+        lrpc_def = load_lrpc_def_from_file(input_file, warnings_as_errors)
         generate_rpc(lrpc_def, output)
+        logging.info("Generated LRPC code for %s in %s", input_file.name, output)
+
+    # catching general exception here is considered ok, because application will terminate
+    # pylint: disable=broad-exception-caught
+    except Exception as e:
+        logging.error("Error while generating code for %s", input_file.name)
+        logging.error(str(e))
 
 
 if __name__ == "__main__":

package/lrpc/lrpcc.py

@@ -1,18 +1,17 @@
 """LRPC client CLI"""
 
+import logging
 import os
 import sys
 from glob import glob
 from os import path
 from typing import Any, Optional
 from collections.abc import Callable
-
 import click
 import serial
 import yaml
 from .client import ClientCliVisitor, LrpcClient
-from .core.definition import LrpcDef
-from .utils import load_lrpc_def
+from .utils import load_lrpc_def_from_url
 
 LRPCC_CONFIG_ENV_VAR = "LRPCC_CONFIG"
 LRPCC_CONFIG_YAML = "lrpcc.config.yaml"
@@ -104,8 +103,10 @@ def create(definition_url: str, transport_type: str) -> None:
 # pylint: disable = too-few-public-methods
 class Lrpcc:
     def __init__(self, config: dict[str, Any]) -> None:
-        self.lrpc_def: LrpcDef = load_lrpc_def(config[DEFINITION_URL])
-        self.client = LrpcClient(config[DEFINITION_URL])
+        input_file = config[DEFINITION_URL]
+        self.lrpc_def = load_lrpc_def_from_url(input_file, warnings_as_errors=True)
+
+        self.client = LrpcClient(self.lrpc_def)
         self.transport_type: str = config[TRANSPORT_TYPE]
         self.transport_params: dict[str, Any] = config[TRANSPORT_PARAMS]
 
@@ -141,10 +142,18 @@ def run(self) -> None:
 def run_cli() -> None:
     config = __load_config()
 
-    if config:
-        Lrpcc(config).run()
-    else:
+    if not config:
         run_lrpcc_config_creator()
+        return
+
+    try:
+        Lrpcc(config).run()
+
+    # catching general exception here is considered ok, because application will terminate
+    # pylint: disable=broad-exception-caught
+    except Exception as e:
+        logging.error("Error running LRPC command line client for %s", config[DEFINITION_URL])
+        logging.error(str(e))
 
 
 if __name__ == "__main__":

package/lrpc/schema/__init__.py

@@ -1 +1 @@
-from .load import load_lrpc_schema
+from .load import load_lrpc_schema as load_lrpc_schema

package/lrpc/utils/__init__.py

@@ -1 +1,4 @@
-from .load_definition import load_lrpc_def
+from .load_definition import load_lrpc_def_from_url as load_lrpc_def_from_url
+from .load_definition import load_lrpc_def_from_str as load_lrpc_def_from_str
+from .load_definition import load_lrpc_def_from_dict as load_lrpc_def_from_dict
+from .load_definition import load_lrpc_def_from_file as load_lrpc_def_from_file