Replace mypy with ty type checker and fix all type issues in src/ and tests/ directories

.pre-commit-config.yaml

@@ -49,6 +49,17 @@ repos:
   - id: check-merge-conflict
   - id: trailing-whitespace
 
+- repo: local
+  hooks:
+  - id: ty
+    name: ty
+    entry: ty check src/ tests/
+    language: python
+    types: [ python ]
+    pass_filenames: false
+    additional_dependencies: [ "ty==0.0.8" ]
+    exclude: ^doc/
+
 # TO BE ACTIVATED EVENTUALLY - EXISTING ERRORS SHOULD BE SOLVED
 #
 # - repo: https://github.com/PyCQA/bandit

pyproject.toml

@@ -88,7 +88,7 @@ dev = [
     #    "pip-audit",
     #    "bandit[toml]",
     "codespell",
-    #    "mypy",
+    "ty",
     #    "vulture",
     #    "deptry",
 ]
@@ -201,6 +201,37 @@ ignore-words = "doc/styles/Vocab/ANSYS/accept.txt"
 skip = '*.pyc,*.xml,*.gif,*.png,*.jpg,*.js,*.html,doc/source/examples/**/*.ipynb,*.json,*.gz'
 quiet-level = 3
 
+[tool.ty]
+# Configuration for ty type checker
+#
+# Note: ty vs mypy
+# ---------------
+# In general, if you run mypy in strict mode, then running ty without changing
+# any of its settings is very similar. However, ty currently does not highlight
+# missing type hints. If you need to enforce adding type hints, you can use
+# Ruff's flake8-annotations.
+#
+# Using Flake8 annotations in Ruff:
+# To enable flake8-annotations in your Ruff configuration, add "ANN" to your
+# selected rules list. If you have other rules already selected, you can add
+# "ANN" to the end of the list to enable it.
+
+[tool.ty.environment]
+# Python version to use for type checking
+python-version = "3.10"  # Minimum version from project.requires-python
+
+[tool.ty.rules]
+# Rule configuration for type checking behavior
+# Import and attribute checking
+unresolved-import = "ignore"  # Ignore missing third-party type stubs
+possibly-missing-attribute = "error"  # Fail on potentially missing attributes
+possibly-missing-import = "error"  # Fail on potentially missing imports
+
+[tool.ty.analysis]
+# Respect type: ignore comments (default: true)
+# This matches mypy's behavior with type: ignore comments
+respect-type-ignore-comments = true
+
 [tool.bandit]
 targets = ["src"]
 recursive = true

src/ansys/dynamicreporting/core/adr_report.py

@@ -70,6 +70,8 @@ def __find_report_obj__(self) -> bool:
             ``False`` otherwise.
         """
         success = False
+        assert self.service is not None
+        assert self.service.serverobj is not None
         all_reports = self.service.serverobj.get_objects(objtype=report_objects.TemplateREST)
         report_objs = [x for x in all_reports if x.name == self.report_name]
         if len(report_objs) > 0:
@@ -125,12 +127,14 @@ def visualize(self, new_tab: bool = False, filter: str = "", item_filter: str =
             item_filter = filter
         if in_ipynb() and not new_tab:  # pragma: no cover
             iframe = self.get_iframe()
+            assert self.service is not None
             if iframe is None:  # pragma: no cover
                 self.service.logger.error("Error: can not obtain IFrame for report")
             else:
-                display(iframe)
+                display(iframe)  # type: ignore[name-defined]
         else:
             url = self.get_url(item_filter=item_filter)
+            assert self.service is not None
             if url == "":  # pragma: no cover
                 self.service.logger.error("Error: could not obtain url for report")
             else:
@@ -220,8 +224,8 @@ def get_guid(self) -> str:
         """
         guid = ""
         if self.service is None:  # pragma: no cover
-            self.service.logger.error("No connection to any report")
             return guid
+        assert self.service.logger is not None
         if self.service.serverobj is None or self.service.url is None:  # pragma: no cover
             self.service.logger.error("No connection to any server")
             return guid
@@ -649,11 +653,11 @@ def export_pdf(
         """
         success = False  # pragma: no cover
         if self.service is None:  # pragma: no cover
-            self.service.logger.error("No connection to any report")
-            return ""
+            return False  # type: ignore[return-value]
+        assert self.service.logger is not None
         if self.service.serverobj is None:  # pragma: no cover
-            self.service.logger.error("No connection to any server")
-            return ""
+            self.service.logger.error("No connection to any service")
+            return False  # type: ignore[return-value]
         try:  # pragma: no cover
             if query_params is None:
                 query_params = {}
@@ -717,11 +721,11 @@ def export_html(
         """
         success = False
         if self.service is None:  # pragma: no cover
-            self.service.logger.error("No connection to any report")
-            return ""
+            return False  # type: ignore[return-value]
+        assert self.service.logger is not None
         if self.service.serverobj is None:  # pragma: no cover
             self.service.logger.error("No connection to any server")
-            return ""
+            return False  # type: ignore[return-value]
         try:
             if query_params is None:
                 query_params = {}
@@ -764,8 +768,12 @@ def export_json(self, json_file_path: str) -> None:
             report.export_json(r'C:\\tmp\\my_json_file.json')
         """
         try:
+            assert self.service is not None
+            assert self.service.serverobj is not None
             self.service.serverobj.store_json(self.report.guid, json_file_path)
         except Exception as e:
+            assert self.service is not None
+            assert self.service.logger is not None
             self.service.logger.error(
                 f"Exporting to JSON terminated for report: {self.report_name}\n"
                 f"Error details: {e}"

src/ansys/dynamicreporting/core/adr_service.py

@@ -109,12 +109,12 @@ class Service:
 
     def __init__(
         self,
-        ansys_version: int = None,
+        ansys_version: int | None = None,
         docker_image: str = DOCKER_REPO_URL,
-        data_directory: str = None,
-        db_directory: str = None,
+        data_directory: str | None = None,
+        db_directory: str | None = None,
         port: int = DOCKER_DEFAULT_PORT,
-        logfile: str = None,
+        logfile: str | None = None,
         ansys_installation: str | None = None,
     ) -> None:
         """
@@ -207,7 +207,7 @@ def __init__(
                     host_directory=self._data_directory,
                     db_directory=self._db_directory,
                     port=self._port,
-                    ansys_version=ansys_version,
+                    ansys_version=ansys_version,  # type: ignore[arg-type]
                 )
             except Exception as e:  # pragma: no cover
                 self.logger.error(f"Error starting the Docker Container.\n{str(e)}\n")
@@ -458,7 +458,7 @@ def start(
                 "exec_basis": self._ansys_installation,
                 "ansys_version": self._ansys_version,
             }
-            if int(self._ansys_version) >= 231:
+            if self._ansys_version and int(self._ansys_version) >= 231:  # type: ignore[arg-type]
                 launch_kwargs.update({"allow_iframe_embedding": True})
 
             try:
@@ -510,6 +510,7 @@ def stop(self) -> None:
 
         v = False
         try:
+            assert self.serverobj is not None
             v = self.serverobj.validate()
         except Exception as e:
             self.logger.error(f"Error: {str(e)}")
@@ -525,6 +526,7 @@ def stop(self) -> None:
                     self._docker_launcher = None
                 else:
                     self.logger.info("Shutting down service.\n")
+                    assert self.serverobj is not None
                     self.serverobj.stop_local_server()
             except Exception as e:
                 self.logger.error(f"Problem shutting down container/service.\n{str(e)}\n")
@@ -619,6 +621,7 @@ def visualize_report(
         if self.serverobj is None:
             self.logger.error("No connection to any service")
             raise ConnectionToServiceError
+        assert self._url is not None
         url = self._url + "/reports/report_display/?"
         if report_name:
             all_reports = self.serverobj.get_objects(objtype=report_objects.TemplateREST)
@@ -635,7 +638,7 @@ def visualize_report(
             query_str = ""
         url += query_str
         if in_ipynb() and not new_tab:
-            display(IFrame(src=url, width=1000, height=800))
+            display(IFrame(src=url, width=1000, height=800))  # type: ignore[name-defined]
         else:
             webbrowser.open_new(url)
 
@@ -717,6 +720,7 @@ def query(
         if valid is False:
             self.logger.warning("Warning: item_filter string is not valid. Will be ignored.")
             item_filter = ""
+        assert self.serverobj is not None
         if query_type == "Item":
             org_queried_items = self.serverobj.get_objects(
                 objtype=report_objects.ItemREST, query=item_filter
@@ -813,6 +817,7 @@ def delete(self, items: list) -> None:
                 )
         # Finally removing from database
         try:
+            assert self.serverobj is not None
             _ = self.serverobj.del_objects(items_to_delete)
         except Exception as e:
             self.logger.warning(f"Error in deleting items: {str(e)}")
@@ -955,6 +960,7 @@ def load_templates(self, json_file_path: str) -> None:
                 break
 
         # 2. Compare with the existing root template(s)
+        assert self.serverobj is not None
         templates = self.serverobj.get_objects(objtype=report_objects.TemplateREST)
         existing_root_names = set()
         for template in templates:
@@ -1007,4 +1013,6 @@ def __checkport__(self):
             self.logger.warning(
                 f"Warning: port {self._port} is already in use. Replace with a new port\n"
             )
-            self._port = report_utils.find_unused_ports(count=1, start=self._port)[0]
+            ports = report_utils.find_unused_ports(count=1, start=self._port)
+            assert ports is not None
+            self._port = ports[0]

src/ansys/dynamicreporting/core/common_utils.py

@@ -83,7 +83,7 @@ def get_install_info(
             install_dir = dir_
             break
 
-    version = get_install_version(install_dir)
+    version = get_install_version(install_dir) if install_dir else None
     # use user provided version only if install dir has no version
     if version is None:
         version = ansys_version or int(CURRENT_VERSION)

src/ansys/dynamicreporting/core/docker_support.py

@@ -67,8 +67,8 @@ def __init__(self, image_url: str | None = None, use_dev: bool = False) -> None:
             self._client: docker.client.DockerClient = docker.from_env()
         except Exception as e:  # pragma: no cover
             raise RuntimeError(f"Can't initialize Docker: {str(e)}")
-        self._container: docker.models.containers.Container = None
-        self._image: docker.models.images.Image = None
+        self._container: docker.models.containers.Container = None  # type: ignore[assignment, attr-defined]
+        self._image: docker.models.images.Image = None  # type: ignore[assignment, attr-defined]
         # the Ansys / EnSight version we found in the container
         # to be reassigned later
         self._ansys_version = None
@@ -78,7 +78,7 @@ def __init__(self, image_url: str | None = None, use_dev: bool = False) -> None:
         self._nexus_directory = None
         self._nexus_is_running = False
 
-    def pull_image(self) -> docker.models.images.Image:
+    def pull_image(self) -> docker.models.images.Image:  # type: ignore[attr-defined]
         """
         Ensure the Docker image is available locally.
 
@@ -111,7 +111,7 @@ def pull_image(self) -> docker.models.images.Image:
                 f"Unexpected error while resolving Docker image: {self._image_url}\n\n{str(e)}"
             ) from e
 
-    def create_container(self) -> docker.models.containers.Container:
+    def create_container(self) -> docker.models.containers.Container:  # type: ignore[attr-defined]
         """
         Create a Docker container using the specified image.
         """
@@ -275,7 +275,7 @@ def image(self):
 
         Returns
         -------
-        docker.models.images.Image
+        docker.models.  # type: ignore[attr-defined]images.Image
             Docker image or ``None`` if an image was not found.
         """
         return self._image
@@ -286,7 +286,7 @@ def container(self):
 
         Returns
         -------
-        docker.models.containers.Container
+        docker.models.  # type: ignore[attr-defined]containers.Container
             Docker container or ``None`` if a container was not found.
         """
         return self._container
@@ -304,7 +304,7 @@ def container_name(self) -> str | None:
             return None
         return self._container.name
 
-    def ansys_version(self) -> str:
+    def ansys_version(self) -> str | None:
         """
         Get the three-digit Ansys version from the Docker container.
 
@@ -316,7 +316,7 @@ def ansys_version(self) -> str:
         """
         return self._ansys_version
 
-    def cei_home(self) -> str:
+    def cei_home(self) -> str | None:
         """
         Get the location of the ``CEI_HOME`` directory within the Docker container.
 
@@ -327,7 +327,7 @@ def cei_home(self) -> str:
         """
         return self._cei_home
 
-    def nexus_directory(self) -> str:
+    def nexus_directory(self) -> str | None:
         """
         Get the location of the ``nexusNNN`` directory within the Docker container.
 
@@ -425,6 +425,7 @@ def copy_from_cei_home_to_host_directory(
         cp_cmd = "/bin/cp "
         if do_recursive:
             cp_cmd += "-r "
+        assert self._cei_home is not None
         cp_cmd += self._cei_home + "/" + src
         cp_cmd += " "
         cp_cmd += "/host_directory/"
@@ -445,10 +446,12 @@ def create_nexus_db(self) -> str:
         ------
         RuntimeError
         """
+        assert self._ansys_version is not None
         if int(self._ansys_version) > 242:
             launcher = "adr_launcher"
         else:
             launcher = "nexus_launcher"
+        assert self._cei_home is not None
         nexus_cmd = self._cei_home + f"/bin/{launcher} create --db_directory /db_directory/ "
         return self.run_in_container(nexus_cmd)
 
@@ -467,10 +470,12 @@ def save_config(self) -> str:
         ------
         RuntimeError
         """
+        assert self._ansys_version is not None
         if int(self._ansys_version) > 242:
             launcher = "adr_launcher"
         else:
             launcher = "nexus_launcher"
+        assert self._cei_home is not None
         nexus_cmd = self._cei_home + f"/bin/{launcher}"
         nexus_cmd += " --db_directory /db_directory"
         nexus_cmd += " save_config"
@@ -503,7 +508,9 @@ def launch_nexus_server(
         ------
         RuntimeError
         """
+        assert self._ansys_version is not None
         launcher = "adr_launcher" if int(self._ansys_version) > 242 else "nexus_launcher"
+        assert self._cei_home is not None
         nexus_cmd = (
             f"{self._cei_home}/bin/{launcher} start "
             f"--db_directory /db_directory "
@@ -520,10 +527,12 @@ def stop(self) -> None:
         """Release any additional resources allocated during launching."""
         try:
             if self._nexus_is_running:
+                assert self._ansys_version is not None
                 if int(self._ansys_version) > 242:
                     launcher = "adr_launcher"
                 else:
                     launcher = "nexus_launcher"
+                assert self._cei_home is not None
                 stop_cmd = self._cei_home + f"/bin/{launcher} stop "
                 stop_cmd += " --db_directory /db_directory"
                 self.run_in_container(stop_cmd)

src/ansys/dynamicreporting/core/examples/downloads.py

@@ -104,7 +104,7 @@ def _retrieve_file(
     logging.debug(f'Downloading URL: "{url}"')
     content = get_url_content(url)
     with open(local_path, "wb") as f:
-        f.write(content)
+        f.write(content)  # type: ignore[arg-type]
 
     logging.debug("Download successful.")
     return local_path