chore: add pyright

Author: cdragos

.dockerignore

@@ -1 +1,20 @@
+# Python cache files
+__pycache__
+*.pyc
+
+# IDE settings
+.vscode/
+
+# Version control
+.git/
+
+# Distribution / packaging
+build/
+dist/
+*.egg-info/
+
+# Virtual environments
+.venv
+
+# Testing
 /tests/manual/data

.github/workflows/test.yml

@@ -27,10 +27,8 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install ".[dev]"
-      - name: 🧹 Lint
+      - name: 🧹 Check code quality
         run: |
           make check_code_quality
-      - name: Check types with mypy
-        run: mypy .
-      - name: 🧪 Test
+      - name: 🧪 Run tests
         run: "python -m unittest"

Makefile

@@ -1,4 +1,4 @@
-.PHONY: style check_code_quality
+.PHONY: style check_code_quality publish
 
 export PYTHONPATH = .
 check_dirs := roboflow
@@ -10,6 +10,7 @@ style:
 check_code_quality:
 	ruff format $(check_dirs) --check
 	ruff check $(check_dirs)
+	pyright $(check_dirs)
 
 publish:
 	python setup.py sdist bdist_wheel

pyproject.toml

@@ -11,9 +11,7 @@ target-version = "py38"
 line-length = 120
 
 [tool.ruff.lint]
-select = [
-    "ALL",
-]
+select = ["ALL"]
 ignore = [
     "A",
     "ANN",
@@ -94,23 +92,11 @@ convention = "google"
 # Preserve types, even if a file imports `from __future__ import annotations`.
 keep-runtime-typing = true
 
-[tool.mypy]
-python_version = "3.8"
-exclude = ["^build/"]
-
-[[tool.mypy.overrides]]
-module = [
-    "_datetime.*",
-    "filetype.*",
-    # IPython is an optional dependency
-    "IPython.display.*",
-    # ipywidgets is an optional dependency
-    "ipywidgets.*",
-    # matplotlib typing is not available for Python 3.8
-    # remove this when we stop supporting Python 3.8
-    "matplotlib.*",
-    "requests_toolbelt.*",
-    "torch.*",
-    "ultralytics.*",
-]
-ignore_missing_imports = true
+[tool.pyright]
+pythonVersion = "3.8"
+exclude = ["build", ".venv", "venv*", "docs", "Tests"]
+venvPath = "."
+venv = ".venv"
+reportMissingImports = false
+reportUnboundVariable = false
+reportIncompatibleMethodOverride = false

roboflow/__init__.py

@@ -59,20 +59,13 @@ def check_key(api_key, model, notebook, num_retries=0):
         return "onboarding"
 
 
-def auth(api_key):
-    r = check_key(api_key)
-    w = r["workspace"]
-
-    return Roboflow(api_key, w)
-
-
 def login(workspace=None, force=False):
     os_name = os.name
 
     if os_name == "nt":
-        default_path = os.path.join(os.getenv("USERPROFILE"), "roboflow/config.json")
+        default_path = os.path.join(os.getenv("USERPROFILE", ""), "roboflow/config.json")
     else:
-        default_path = os.path.join(os.getenv("HOME"), ".config/roboflow/config.json")
+        default_path = os.path.join(os.getenv("HOME", ""), ".config/roboflow/config.json")
 
     # default configuration location
     conf_location = os.getenv(
@@ -143,7 +136,7 @@ def initialize_roboflow(the_workspace=None):
 
     conf_location = os.getenv(
         "ROBOFLOW_CONFIG_DIR",
-        default=os.getenv("HOME") + "/.config/roboflow/config.json",
+        default=os.getenv("HOME", "") + "/.config/roboflow/config.json",
     )
 
     if not os.path.isfile(conf_location):
@@ -176,7 +169,7 @@ def load_model(model_url):
         project = path_parts[2]
         version = int(path_parts[-1])
     else:
-        raise ("Model URL must be from either app.roboflow.com or universe.roboflow.com")
+        raise ValueError("Model URL must be from either app.roboflow.com or universe.roboflow.com")
 
     project = operate_workspace.project(project)
     version = project.version(version)
@@ -204,7 +197,7 @@ def download_dataset(dataset_url, model_format, location=None):
         version = int(path_parts[-1])
         the_workspace = path_parts[1]
     else:
-        raise ("Model URL must be from either app.roboflow.com or universe.roboflow.com")
+        raise ValueError("Model URL must be from either app.roboflow.com or universe.roboflow.com")
     operate_workspace = initialize_roboflow(the_workspace=the_workspace)
 
     project = operate_workspace.project(project)
@@ -239,7 +232,7 @@ def auth(self):
             self.universe = True
             return self
         else:
-            w = r["workspace"]
+            w = r["workspace"]  # type: ignore[arg-type]
             self.current_workspace = w
             return self
 
@@ -282,7 +275,7 @@ def project(self, project_name, the_workspace=None):
 
         dataset_info = dataset_info.json()["project"]
 
-        return Project(self.api_key, dataset_info)
+        return Project(self.api_key or "", dataset_info)
 
     def __str__(self):
         """to string function"""

roboflow/adapters/rfapi.py

@@ -1,7 +1,7 @@
 import json
 import os
-import urllib
 from typing import Optional
+from urllib import parse
 
 import requests
 from requests_toolbelt.multipart.encoder import MultipartEncoder
@@ -164,7 +164,7 @@ def _upload_url(api_key, project_url, **kwargs):
     url = f"{API_URL}/dataset/{project_url}/upload?api_key={api_key}"
 
     if kwargs:
-        querystring = urllib.parse.urlencode(kwargs, doseq=True)
+        querystring = parse.urlencode(kwargs, doseq=True)
         url += f"&{querystring}"
 
     return url

roboflow/config.py

@@ -1,8 +1,9 @@
 import json
 import os
+from typing import Any
 
 
-def get_conditional_configuration_variable(key, default):
+def get_conditional_configuration_variable(key, default) -> Any:
     """Retrieves the configuration variable conditionally.
         ##1. check if variable is in environment
         ##2. check if variable is in config file
@@ -17,9 +18,9 @@ def get_conditional_configuration_variable(key, default):
     os_name = os.name
 
     if os_name == "nt":
-        default_path = os.path.join(os.getenv("USERPROFILE"), "roboflow/config.json")
+        default_path = os.path.join(os.getenv("USERPROFILE", ""), "roboflow/config.json")
     else:
-        default_path = os.path.join(os.getenv("HOME"), ".config/roboflow/config.json")
+        default_path = os.path.join(os.getenv("HOME", ""), ".config/roboflow/config.json")
 
     # default configuration location
     conf_location = os.getenv(
@@ -34,9 +35,10 @@ def get_conditional_configuration_variable(key, default):
     else:
         config = {}
 
-    if os.getenv(key) is not None:
-        return os.getenv(key)
-    elif key in config.keys():
+    env_value = os.getenv(key)
+    if env_value is not None:
+        return env_value
+    elif key in config:
         return config[key]
     else:
         return default

roboflow/core/project.py

@@ -230,7 +230,7 @@ def generate_version(self, settings):
         try:
             r_json = r.json()
         except Exception:
-            raise "Error when requesting to generate a new version for project."
+            raise RuntimeError("Error when requesting to generate a new version for project.")
 
         # if the generation succeeds, return the version that is being generated
         if r.status_code == 200:
@@ -256,7 +256,7 @@ def train(
         speed=None,
         checkpoint=None,
         plot_in_notebook=False,
-    ) -> bool:
+    ):
         """
         Ask the Roboflow API to train a previously exported version's dataset.
 
@@ -503,7 +503,7 @@ def single_upload(
                     sequence_size=sequence_size,
                     **kwargs,
                 )
-                image_id = uploaded_image["id"]
+                image_id = uploaded_image["id"]  # type: ignore[index]
                 upload_retry_attempts = retry.retries
             except BaseException as e:
                 uploaded_image = {"error": e}
@@ -518,10 +518,10 @@ def single_upload(
                 uploaded_annotation = rfapi.save_annotation(
                     self.__api_key,
                     project_url,
-                    annotation_name,
-                    annotation_str,
+                    annotation_name,  # type: ignore[type-var]
+                    annotation_str,  # type: ignore[type-var]
                     image_id,
-                    job_name=batch_name,
+                    job_name=batch_name,  # type: ignore[type-var]
                     is_prediction=is_prediction,
                     annotation_labelmap=annotation_labelmap,
                     overwrite=annotation_overwrite,
@@ -543,10 +543,10 @@ def _annotation_params(self, annotation_path):
         if isinstance(annotation_path, dict) and annotation_path.get("rawText"):
             annotation_name = annotation_path["name"]
             annotation_string = annotation_path["rawText"]
-        elif os.path.exists(annotation_path):
-            with open(annotation_path):
-                annotation_string = open(annotation_path).read()
-            annotation_name = os.path.basename(annotation_path)
+        elif os.path.exists(annotation_path):  # type: ignore[arg-type]
+            with open(annotation_path):  # type: ignore[arg-type]
+                annotation_string = open(annotation_path).read()  # type: ignore[arg-type]
+            annotation_name = os.path.basename(annotation_path)  # type: ignore[arg-type]
         elif self.type == "classification":
             print(f"-> using {annotation_path} as classname for classification project")
             annotation_string = annotation_path

roboflow/core/version.py

@@ -766,7 +766,7 @@ def bar_progress(current, total, width=80):
 
             # write the zip file to the desired location
             with open(location + "/roboflow.zip", "wb") as f:
-                total_length = int(response.headers.get("content-length"))
+                total_length = int(response.headers.get("content-length"))  # type: ignore[arg-type]
                 desc = None if TQDM_DISABLE else f"Downloading Dataset Version Zip in {location} to {format}:"
                 for chunk in tqdm(
                     response.iter_content(chunk_size=1024),

roboflow/core/workspace.py

@@ -3,7 +3,7 @@
 import json
 import os
 import sys
-from typing import List
+from typing import Any, List
 
 import numpy as np
 import requests
@@ -179,7 +179,7 @@ def two_stage(
         print(self.project(first_stage_model_name))
 
         # perform first inference
-        predictions = stage_one_model.predict(image)
+        predictions = stage_one_model.predict(image)  # type: ignore[attribute-error]
 
         if stage_one_project.type == "object-detection" and stage_two_project == "classification":
             # interact with each detected object from stage one inference results
@@ -199,7 +199,7 @@ def two_stage(
                 croppedImg.save("./temp.png")
 
                 # capture results of second stage inference from cropped image
-                results.append(stage_two_model.predict("./temp.png")[0])
+                results.append(stage_two_model.predict("./temp.png")[0])  # type: ignore[attribute-error]
 
             # delete the written image artifact
             try:
@@ -244,7 +244,7 @@ def two_stage_ocr(
         stage_one_model = stage_one_project.version(first_stage_model_version).model
 
         # perform first inference
-        predictions = stage_one_model.predict(image)
+        predictions = stage_one_model.predict(image)  # type: ignore[attribute-error]
 
         # interact with each detected object from stage one inference results
         if stage_one_project.type == "object-detection":
@@ -391,7 +391,7 @@ def active_learning(
         upload_destination: str = "",
         conditionals: dict = {},
         use_localhost: bool = False,
-    ) -> str:
+    ) -> Any:
         """perform inference on each image in directory and upload based on conditions
         @params:
             raw_data_location: (str) = folder of frames to be processed
@@ -470,7 +470,7 @@ def active_learning(
                     print(image2 + " --> similarity too high to --> " + image1)
                     continue  # skip this image if too similar or counter hits limit
 
-            predictions = inference_model.predict(image).json()["predictions"]
+            predictions = inference_model.predict(image).json()["predictions"]  # type: ignore[attribute-error]
             # collect all predictions to return to user at end
             prediction_results.append({"image": image, "predictions": predictions})
 

roboflow/models/classification.py

@@ -2,8 +2,8 @@
 import io
 import json
 import os
-import urllib
 from typing import Optional
+from urllib import parse
 
 import requests
 from PIL import Image
@@ -106,7 +106,7 @@ def predict(self, image_path, hosted=False):
             )
         else:
             # Create API URL for hosted image (slightly different)
-            self.api_url += "&image=" + urllib.parse.quote_plus(image_path)
+            self.api_url += "&image=" + parse.quote_plus(image_path)
             # POST to the API
             resp = requests.post(self.api_url)
             img_dims = {"width": "0", "height": "0"}

roboflow/models/inference.py

@@ -2,8 +2,8 @@
 import json
 import os
 import time
-import urllib
 from typing import Optional, Tuple
+from urllib import parse
 from urllib.parse import urljoin
 
 import requests
@@ -72,7 +72,7 @@ def __get_image_params(self, image_path):
         """
         validate_image_path(image_path)
 
-        hosted_image = urllib.parse.urlparse(image_path).scheme in ("http", "https")
+        hosted_image = parse.urlparse(image_path).scheme in ("http", "https")
 
         if hosted_image:
             image_dims = {"width": "Undefined", "height": "Undefined"}
@@ -121,8 +121,7 @@ def predict(self, image_path, prediction_type=None, **kwargs):
         params["api_key"] = self.__api_key
 
         params.update(**kwargs)
-
-        url = f"{self.api_url}?{urllib.parse.urlencode(params)}"
+        url = f"{self.api_url}?{parse.urlencode(params)}"  # type: ignore[attr-defined]
         response = requests.post(url, **request_kwargs)
         response.raise_for_status()
 
@@ -390,7 +389,7 @@ def download(self, format="pt", location="."):
 
         # write the zip file to the desired location
         with open(location + "/weights.pt", "wb") as f:
-            total_length = int(response.headers.get("content-length"))
+            total_length = int(response.headers.get("content-length"))  # type: ignore[arg-type]
             for chunk in tqdm(
                 response.iter_content(chunk_size=1024),
                 desc=f"Downloading weights to {location}/weights.pt",