Merge branch 'master' into video-inference

Author: David Eigen

CHANGELOG.md

@@ -1,3 +1,22 @@
+## [[11.1.1]](https://github.com/Clarifai/clarifai-python/releases/tag/11.1.1) - [PyPI](https://pypi.org/project/clarifai/11.1.1/) - 2025-02-06
+
+### Changed
+
+ - Don't validate API in server.py [(#509)] (https://github.com/Clarifai/clarifai-python/pull/509)
+
+## [[11.1.0]](https://github.com/Clarifai/clarifai-python/releases/tag/11.1.0) - [PyPI](https://pypi.org/project/clarifai/11.1.0/) - 2025-02-05
+
+### Changed
+
+ - Fixed Docker test locally [(#505)] (https://github.com/Clarifai/clarifai-python/pull/505)
+ - Fixed HF checkpoints error [(#504)] (https://github.com/Clarifai/clarifai-python/pull/504)
+ - Fixed Deployment Tests [(#502)] (https://github.com/Clarifai/clarifai-python/pull/502)
+ - Fixed Issue with Filename as Invalid Input ID [(#501)] (https://github.com/Clarifai/clarifai-python/pull/501)
+ - Update Model Predict CLI [(#500)] (https://github.com/Clarifai/clarifai-python/pull/500)
+ - Tests Health Port to None [(#499)] (https://github.com/Clarifai/clarifai-python/pull/499)
+ - Refactor model class and runners to be more independent [(#494)] (https://github.com/Clarifai/clarifai-python/pull/494)
+ - Add storage request inferred from tar and checkpoint size [(#479)] (https://github.com/Clarifai/clarifai-python/pull/479)
+
 ## [[11.0.7]](https://github.com/Clarifai/clarifai-python/releases/tag/11.0.7) - [PyPI](https://pypi.org/project/clarifai/11.0.7/) - 2025-01-24
 
 ### Changed

clarifai/__init__.py

@@ -1 +1 @@
-__version__ = "11.0.7"
+__version__ = "11.1.1"

clarifai/cli/model.py

@@ -172,8 +172,6 @@ def run_locally(model_path, port, mode, keep_env, keep_image):
 @click.option('--file_path', required=False, help='File path of file for the model to predict')
 @click.option('--url', required=False, help='URL to the file for the model to predict')
 @click.option('--bytes', required=False, help='Bytes to the file for the model to predict')
-@click.option(
-    '--input_id', required=False, help='Existing input id in the app for the model to predict')
 @click.option('--input_type', required=False, help='Type of input')
 @click.option(
     '-cc_id',
@@ -187,36 +185,28 @@ def run_locally(model_path, port, mode, keep_env, keep_image):
     '--inference_params', required=False, default='{}', help='Inference parameters to override')
 @click.option('--output_config', required=False, default='{}', help='Output config to override')
 @click.pass_context
-def predict(ctx, config, model_id, user_id, app_id, model_url, file_path, url, bytes, input_id,
-            input_type, compute_cluster_id, nodepool_id, deployment_id, inference_params,
-            output_config):
+def predict(ctx, config, model_id, user_id, app_id, model_url, file_path, url, bytes, input_type,
+            compute_cluster_id, nodepool_id, deployment_id, inference_params, output_config):
   """Predict using the given model"""
   import json
 
-  from clarifai.client.deployment import Deployment
-  from clarifai.client.input import Input
   from clarifai.client.model import Model
-  from clarifai.client.nodepool import Nodepool
   from clarifai.utils.cli import from_yaml
   if config:
     config = from_yaml(config)
-    model_id, user_id, app_id, model_url, file_path, url, bytes, input_id, input_type, compute_cluster_id, nodepool_id, deployment_id, inference_params, output_config = (
+    model_id, user_id, app_id, model_url, file_path, url, bytes, input_type, compute_cluster_id, nodepool_id, deployment_id, inference_params, output_config = (
         config.get(k, v)
         for k, v in [('model_id', model_id), ('user_id', user_id), ('app_id', app_id), (
             'model_url', model_url), ('file_path', file_path), ('url', url), ('bytes', bytes), (
-                'input_id',
-                input_id), ('input_type',
-                            input_type), ('compute_cluster_id',
-                                          compute_cluster_id), ('nodepool_id', nodepool_id), (
-                                              'deployment_id',
-                                              deployment_id), ('inference_params',
-                                                               inference_params), ('output_config',
-                                                                                   output_config)])
+                'input_type', input_type), ('compute_cluster_id', compute_cluster_id), (
+                    'nodepool_id',
+                    nodepool_id), ('deployment_id',
+                                   deployment_id), ('inference_params',
+                                                    inference_params), ('output_config',
+                                                                        output_config)])
   if sum([opt[1] for opt in [(model_id, 1), (user_id, 1), (app_id, 1), (model_url, 3)]
           if opt[0]]) != 3:
     raise ValueError("Either --model_id & --user_id & --app_id or --model_url must be provided.")
-  if sum([1 for opt in [file_path, url, bytes, input_id] if opt]) != 1:
-    raise ValueError("Exactly one of --file_path, --url, --bytes or --input_id must be provided.")
   if compute_cluster_id or nodepool_id or deployment_id:
     if sum([
         opt[1] for opt in [(compute_cluster_id, 0.5), (nodepool_id, 0.5), (deployment_id, 1)]
@@ -266,21 +256,5 @@ def predict(ctx, config, model_id, user_id, app_id, model_url, file_path, url, b
         nodepool_id=nodepool_id,
         deployment_id=deployment_id,
         inference_params=inference_params,
-        output_config=output_config)
-  elif input_id:
-    inputs = [Input.get_input(input_id)]
-    runner_selector = None
-    if deployment_id:
-      runner_selector = Deployment.get_runner_selector(
-          user_id=ctx.obj['user_id'], deployment_id=deployment_id)
-    elif compute_cluster_id and nodepool_id:
-      runner_selector = Nodepool.get_runner_selector(
-          user_id=ctx.obj['user_id'],
-          compute_cluster_id=compute_cluster_id,
-          nodepool_id=nodepool_id)
-    model_prediction = model.predict(
-        inputs=inputs,
-        runner_selector=runner_selector,
-        inference_params=inference_params,
-        output_config=output_config)
+        output_config=output_config)  ## TO DO: Add support for input_id
   click.echo(model_prediction)

clarifai/client/input.py

@@ -572,6 +572,24 @@ def get_mask_proto(input_id: str,
 
     return input_mask_proto
 
+  def get_input(self, input_id: str) -> Input:
+    """Get Input object of input with input_id provided from the app.
+
+    Args:
+        input_id (str): The input ID for the annotation to get.
+
+    Returns:
+        Input: An Input object for the specified input ID.
+
+    Example:
+        >>> from clarifai.client.input import Inputs
+        >>> input_obj = Inputs(user_id = 'user_id', app_id = 'demo_app')
+        >>> input_obj.get_input(input_id='demo')
+    """
+    request = service_pb2.GetInputRequest(user_app_id=self.user_app_id, input_id=input_id)
+    response = self._grpc_request(self.STUB.GetInput, request)
+    return response.input
+
   def upload_from_url(self,
                       input_id: str,
                       image_url: str = None,

clarifai/runners/dockerfile_template/Dockerfile.template

@@ -73,7 +73,7 @@ ENV PYTHONPATH=${PYTHONPATH}:/home/nonroot/main \
     CLARIFAI_RUNNER_ID=${CLARIFAI_RUNNER_ID} \
     CLARIFAI_NODEPOOL_ID=${CLARIFAI_NODEPOOL_ID} \
     CLARIFAI_COMPUTE_CLUSTER_ID=${CLARIFAI_COMPUTE_CLUSTER_ID} \
-    CLARIFAI_API_BASE=${CLARIFAI_API_BASE}
+    CLARIFAI_API_BASE=${CLARIFAI_API_BASE:-https://api.clarifai.com}
 
 # Finally run the clarifai entrypoint to start the runner loop and local dev server.
 # Note(zeiler): we may want to make this a clarifai CLI call.

clarifai/runners/models/model_builder.py

@@ -32,6 +32,7 @@ def _clear_line(n: int = 1) -> None:
 
 
 class ModelBuilder:
+  DEFAULT_CHECKPOINT_SIZE = 50 * 1024**3  # 50 GiB
 
   def __init__(self, folder: str, validate_api_ids: bool = True, download_validation_only=False):
     """
@@ -154,6 +155,9 @@ def _check_app_exists(self):
     resp = self.client.STUB.GetApp(service_pb2.GetAppRequest(user_app_id=self.client.user_app_id))
     if resp.status.code == status_code_pb2.SUCCESS:
       return True
+    logger.error(
+        f"Error checking API {self._base_api} for user app {self.client.user_app_id.user_id}/{self.client.user_app_id.app_id}. Error code: {resp.status.code}"
+    )
     return False
 
   def _validate_config_model(self):
@@ -200,6 +204,24 @@ def _validate_config(self):
           )
           logger.info("Continuing without Hugging Face token")
 
+  @staticmethod
+  def _get_tar_file_content_size(tar_file_path):
+    """
+    Calculates the total size of the contents of a tar file.
+
+    Args:
+      tar_file_path (str): The path to the tar file.
+
+    Returns:
+      int: The total size of the contents in bytes.
+    """
+    total_size = 0
+    with tarfile.open(tar_file_path, 'r') as tar:
+      for member in tar:
+        if member.isfile():
+          total_size += member.size
+    return total_size
+
   @property
   def client(self):
     if self._client is None:
@@ -211,9 +233,8 @@ def client(self):
       user_id = model.get('user_id')
       app_id = model.get('app_id')
 
-      base = os.environ.get('CLARIFAI_API_BASE', 'https://api.clarifai.com')
-
-      self._client = BaseClient(user_id=user_id, app_id=app_id, base=base)
+      self._base_api = os.environ.get('CLARIFAI_API_BASE', 'https://api.clarifai.com')
+      self._client = BaseClient(user_id=user_id, app_id=app_id, base=self._base_api)
 
     return self._client
 
@@ -520,6 +541,18 @@ def filter_func(tarinfo):
     file_size = os.path.getsize(self.tar_file)
     logger.info(f"Size of the tar is: {file_size} bytes")
 
+    self.storage_request_size = self._get_tar_file_content_size(file_path)
+    if not download_checkpoints and self.config.get("checkpoints"):
+      # Get the checkpoint size to add to the storage request.
+      # First check for the env variable, then try querying huggingface. If all else fails, use the default.
+      checkpoint_size = os.environ.get('CHECKPOINT_SIZE_BYTES', 0)
+      if not checkpoint_size:
+        _, repo_id, _ = self._validate_config_checkpoints()
+        checkpoint_size = HuggingFaceLoader.get_huggingface_checkpoint_total_size(repo_id)
+      if not checkpoint_size:
+        checkpoint_size = self.DEFAULT_CHECKPOINT_SIZE
+      self.storage_request_size += checkpoint_size
+
     self.maybe_create_model()
     if not self.check_model_exists():
       logger.error(f"Failed to create model: {self.model_proto.id}")
@@ -594,6 +627,7 @@ def init_upload_model_version(self, model_version_proto, file_path):
             model_id=self.model_proto.id,
             model_version=model_version_proto,
             total_size=file_size,
+            storage_request_size=self.storage_request_size,
             is_v3=self.is_v3,
         ))
     return result

clarifai/runners/models/model_run_locally.py

@@ -287,7 +287,9 @@ def build_docker_image(
       # Comment out the COPY instruction that copies the current folder
       modified_lines = []
       for line in lines:
-        if 'COPY .' in line and '/app/model_dir/main' in line:
+        if 'COPY' in line and '/home/nonroot/main' in line:
+          modified_lines.append(f'# {line}')
+        elif 'download-checkpoints' in line and '/home/nonroot/main' in line:
           modified_lines.append(f'# {line}')
         else:
           modified_lines.append(line)
@@ -338,15 +340,15 @@ def run_docker_container(self,
       if self._gpu_is_available():
         cmd.extend(["--gpus", "all"])
       # Add volume mappings
-      cmd.extend(["-v", f"{self.model_path}:/app/model_dir/main"])
+      cmd.extend(["-v", f"{self.model_path}:/home/nonroot/main"])
       # Add environment variables
       if env_vars:
         for key, value in env_vars.items():
           cmd.extend(["-e", f"{key}={value}"])
       # Add the image name
       cmd.append(image_name)
       # update the CMD to run the server
-      cmd.extend(["--model_path", "/app/model_dir/main", "--grpc", "--port", str(port)])
+      cmd.extend(["--model_path", "/home/nonroot/main", "--grpc", "--port", str(port)])
       # Run the container
       process = subprocess.Popen(cmd,)
       logger.info(
@@ -387,7 +389,7 @@ def test_model_container(self,
       # update the entrypoint for testing the model
       cmd.extend(["--entrypoint", "python"])
       # Add volume mappings
-      cmd.extend(["-v", f"{self.model_path}:/app/model_dir/main"])
+      cmd.extend(["-v", f"{self.model_path}:/home/nonroot/main"])
       # Add environment variables
       if env_vars:
         for key, value in env_vars.items():
@@ -397,7 +399,7 @@ def test_model_container(self,
       # update the CMD to test the model inside the container
       cmd.extend([
           "-c",
-          "from clarifai.runners.models.model_run_locally import ModelRunLocally; ModelRunLocally('/app/model_dir/main')._run_test()"
+          "from clarifai.runners.models.model_run_locally import ModelRunLocally; ModelRunLocally('/home/nonroot/main')._run_test()"
       ])
       # Run the container
       subprocess.check_call(cmd)

clarifai/runners/server.py

@@ -68,7 +68,7 @@ def main():
 
   parsed_args = parser.parse_args()
 
-  builder = ModelBuilder(parsed_args.model_path)
+  builder = ModelBuilder(parsed_args.model_path, download_validation_only=True)
 
   model = builder.create_model_instance()
 
@@ -101,7 +101,7 @@ def main():
         runner_id=os.environ["CLARIFAI_RUNNER_ID"],
         nodepool_id=os.environ["CLARIFAI_NODEPOOL_ID"],
         compute_cluster_id=os.environ["CLARIFAI_COMPUTE_CLUSTER_ID"],
-        base_url=os.environ["CLARIFAI_API_BASE"],
+        base_url=os.environ.get("CLARIFAI_API_BASE", "https://api.clarifai.com"),
         num_parallel_polls=int(os.environ.get("CLARIFAI_NUM_THREADS", 1)),
     )
     runner.start()  # start the runner to fetch work from the API.

clarifai/runners/utils/loader.py

@@ -4,6 +4,8 @@
 import os
 import shutil
 
+import requests
+
 from clarifai.utils.logging import logger
 
 
@@ -23,14 +25,14 @@ def __init__(self, repo_id=None, token=None):
         login(token=token)
         logger.info("Hugging Face token validated")
       else:
+        self.token = None
         logger.info("Continuing without Hugging Face token")
 
   @classmethod
   def validate_hftoken(cls, hf_token: str):
     try:
       if importlib.util.find_spec("huggingface_hub") is None:
         raise ImportError(cls.HF_DOWNLOAD_TEXT)
-      os.environ['HF_TOKEN'] = hf_token
       from huggingface_hub import HfApi
 
       api = HfApi()
@@ -169,3 +171,34 @@ def fetch_labels(checkpoint_path: str):
 
     labels = config['id2label']
     return labels
+
+  @staticmethod
+  def get_huggingface_checkpoint_total_size(repo_name):
+    """
+    Fetches the JSON data for a Hugging Face model using the API with `?blobs=true`.
+    Calculates the total size from the JSON output.
+
+    Args:
+        repo_name (str): The name of the model on Hugging Face Hub. e.g. "casperhansen/llama-3-8b-instruct-awq"
+
+    Returns:
+        int: The total size in bytes.
+    """
+    try:
+      url = f"https://huggingface.co/api/models/{repo_name}?blobs=true"
+      response = requests.get(url)
+      response.raise_for_status()  # Raise an exception for bad status codes
+      json_data = response.json()
+
+      if isinstance(json_data, str):
+        data = json.loads(json_data)
+      else:
+        data = json_data
+
+      total_size = 0
+      for file in data['siblings']:
+        total_size += file['size']
+      return total_size
+    except Exception as e:
+      logger.error(f"Error fetching checkpoint size from huggingface.co: {e}")
+      return 0

requirements.txt

@@ -1,4 +1,4 @@
-clarifai-grpc>=11.0.5
+clarifai-grpc>=11.0.7
 clarifai-protocol>=0.0.16
 numpy>=1.22.0
 tqdm>=4.65.0
@@ -9,3 +9,4 @@ Pillow>=9.5.0
 tabulate>=0.9.0
 fsspec==2024.6.1
 click==8.1.7
+requests==2.32.3