feat: Using args as arguments (#230)

Author: vijayvammi

examples/common/functions.py

@@ -1,3 +1,4 @@
+import argparse
 import logging
 import os
 from pathlib import Path
@@ -32,6 +33,15 @@ class ComplexParams(BaseModel):
     foo: str
 
 
+def function_using_argparse(args: argparse.Namespace):
+    assert args.integer == 1
+    assert args.floater == 3.14
+    assert args.stringer == "hello"
+    assert args.envvar == "from env"
+    assert args.pydantic_param["x"] == 10
+    assert args.pydantic_param["foo"] == "bar"
+
+
 def read_initial_params_as_pydantic(
     integer: int,
     floater: float,

examples/torch/single_cpu_args.py

@@ -0,0 +1,109 @@
+# single_cpu_train_with_args.py
+
+import argparse  # New: for command-line arguments
+import time
+
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torch.utils.data import DataLoader, TensorDataset
+
+
+def run_single_cpu_training(args: argparse.Namespace):
+    """
+    Runs a simple training loop on a single CPU core.
+    Accepts parsed arguments for hyperparameters.
+    """
+    print(
+        f"Parameters: learning_rate={args.learning_rate}, num_epochs={args.num_epochs}, batch_size={args.batch_size}"
+    )
+    print("--- Starting Single-CPU Training ---")
+    print(f"Learning Rate: {args.learning_rate}, Epochs: {args.num_epochs}")
+    print(f"Batch Size: {args. batch_size}")
+
+    # 1. Define a simple model
+    class SimpleModel(nn.Module):
+        def __init__(self):
+            super().__init__()
+            self.linear = nn.Linear(10, 1)  # Input 10 features, output 1
+
+        def forward(self, x):
+            return self.linear(x)
+
+    model = SimpleModel()
+    device = torch.device("cpu")  # Explicitly set device to CPU
+    model.to(device)
+
+    # 2. Create a dummy dataset
+    num_samples = 1000  # Larger dataset to see the difference in speed later
+    num_features = 10
+    X = torch.randn(num_samples, num_features)
+    y = (
+        torch.sum(X * torch.arange(1, num_features + 1).float(), dim=1, keepdim=True)
+        + torch.randn(num_samples, 1) * 0.1
+    )
+
+    dataset = TensorDataset(X, y)
+    dataloader = DataLoader(
+        dataset, batch_size=args.batch_size, shuffle=True
+    )  # Use batch_size parameter
+
+    # 3. Define optimizer and loss function
+    optimizer = optim.SGD(
+        model.parameters(), lr=args.learning_rate
+    )  # Use learning_rate parameter
+    criterion = nn.MSELoss()
+
+    start_time = time.time()
+
+    # 4. Training loop
+    for epoch in range(args.num_epochs):  # Use num_epochs parameter
+        model.train()
+        total_loss = 0
+        for batch_idx, (inputs, targets) in enumerate(dataloader):
+            inputs, targets = inputs.to(device), targets.to(device)
+
+            optimizer.zero_grad()
+            outputs = model(inputs)
+            loss = criterion(outputs, targets)
+            loss.backward()
+            optimizer.step()
+            total_loss += loss.item()
+
+        avg_loss = total_loss / len(dataloader)
+        print(f"Epoch {epoch+1}/{args.num_epochs}, Loss: {avg_loss:.4f}")
+
+    end_time = time.time()
+    print(f"\nSingle-CPU Training complete in {end_time - start_time:.2f} seconds!")
+
+    # Save the model
+    model_save_path = (
+        f"single_cpu_model_lr{args.learning_rate}_epochs{args.num_epochs}.pth"
+    )
+    torch.save(model.state_dict(), model_save_path)
+    print(f"Model saved to {model_save_path}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Single-CPU PyTorch Training Example.")
+    parser.add_argument(
+        "--learning_rate",
+        type=float,
+        default=0.01,
+        help="Learning rate for the optimizer (default: 0.01)",
+    )
+    parser.add_argument(
+        "--num_epochs",
+        type=int,
+        default=50,
+        help="Number of training epochs (default: 50)",
+    )
+    parser.add_argument(
+        "--batch_size",
+        type=int,
+        default=32,
+        help="Batch size for training (default: 32)",
+    )
+
+    args = parser.parse_args()
+    run_single_cpu_training(args)

pyproject.toml

@@ -151,7 +151,7 @@ file-system = "extensions.run_log_store.file_system:FileSystemRunLogstore"
 
 # Release configuration
 [tool.semantic_release]
-commit_parser = "angular"
+commit_parser = "conventional"
 major_on_zero = true
 allow_zero_version = true
 tag_format = "{version}"

runnable/parameters.py

@@ -1,10 +1,10 @@
+import argparse
 import inspect
 import json
 import logging
 import os
 from typing import Any, Dict, Type
 
-import pydantic
 from pydantic import BaseModel, ConfigDict
 from typing_extensions import Callable
 
@@ -48,15 +48,40 @@ def get_user_set_parameters(remove: bool = False) -> Dict[str, JsonParameter]:
     return parameters
 
 
+def return_json_parameters(params: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Returns the parameters as a JSON serializable dictionary.
+
+    Args:
+        params (dict): The parameters to serialize.
+
+    Returns:
+        dict: The JSON serializable dictionary.
+    """
+    return_params = {}
+    for key, value in params.items():
+        if isinstance(value, ObjectParameter):
+            continue
+
+        return_params[key] = value.get_value()
+    return return_params
+
+
 def filter_arguments_for_func(
     func: Callable[..., Any],
     params: Dict[str, Any],
     map_variable: MapVariableType = None,
 ) -> Dict[str, Any]:
     """
     Inspects the function to be called as part of the pipeline to find the arguments of the function.
-    Matches the function arguments to the parameters available either by command line or by up stream steps.
+    Matches the function arguments to the parameters available either by static parameters or by up stream steps.
 
+    The function "func" signature could be:
+    - def my_function(arg1: int, arg2: str, arg3: float):
+    - def my_function(arg1: int, arg2: str, arg3: float, **kwargs):
+        in this case, we would need to send in remaining keyword arguments as a dictionary.
+    - def my_function(arg1: int, arg2: str, arg3: float, args: argparse.Namespace):
+        In this case, we need to send the rest of the parameters as attributes of the args object.
 
     Args:
         func (Callable): The function to inspect
@@ -72,63 +97,109 @@ def filter_arguments_for_func(
         params[key] = JsonParameter(kind="json", value=v)
 
     bound_args = {}
-    unassigned_params = set(params.keys())
-    # Check if VAR_KEYWORD is used, it is we send back everything
-    # If **kwargs is present in the function signature, we send back everything
-    for name, value in function_args.items():
-        if value.kind != inspect.Parameter.VAR_KEYWORD:
-            continue
-        # Found VAR_KEYWORD, we send back everything as found
-        for key, value in params.items():
-            bound_args[key] = params[key].get_value()
+    missing_required_args: list[str] = []
+    var_keyword_param = None
+    namespace_param = None
 
-        return bound_args
-
-    # Lets return what is asked for then!!
+    # First pass: Handle regular parameters and identify special parameters
     for name, value in function_args.items():
         # Ignore any *args
         if value.kind == inspect.Parameter.VAR_POSITIONAL:
             logger.warning(f"Ignoring parameter {name} as it is VAR_POSITIONAL")
             continue
 
-        if name not in params:
-            # No parameter of this name was provided
-            if value.default == inspect.Parameter.empty:
-                # No default value is given in the function signature. error as parameter is required.
-                raise ValueError(
-                    f"Parameter {name} is required for {func.__name__} but not provided"
-                )
-            # default value is given in the function signature, nothing further to do.
+        # Check for **kwargs parameter
+        if value.kind == inspect.Parameter.VAR_KEYWORD:
+            var_keyword_param = name
             continue
 
-        param_value = params[name]
-
-        if type(value.annotation) in [
-            BaseModel,
-            pydantic._internal._model_construction.ModelMetaclass,
-        ] and not isinstance(param_value, ObjectParameter):
-            # Even if the annotation is a pydantic model, it can be passed as an object parameter
-            # We try to cast it as a pydantic model if asked
-            named_param = params[name].get_value()
-
-            if not isinstance(named_param, dict):
-                # A case where the parameter is a one attribute model
-                named_param = {name: named_param}
-
-            bound_model = bind_args_for_pydantic_model(named_param, value.annotation)
-            bound_args[name] = bound_model
+        # Check for argparse.Namespace parameter
+        if value.annotation == argparse.Namespace:
+            namespace_param = name
+            continue
 
-        elif value.annotation in [str, int, float, bool]:
-            # Cast it if its a primitive type. Ensure the type matches the annotation.
-            bound_args[name] = value.annotation(params[name].get_value())
+        # Handle regular parameters
+        if name not in params:
+            if value.default != inspect.Parameter.empty:
+                # Default value is given in the function signature, we can use it
+                bound_args[name] = value.default
+            else:
+                # This is a required parameter that's missing
+                missing_required_args.append(name)
         else:
-            bound_args[name] = params[name].get_value()
-
-        unassigned_params.remove(name)
-
-        params = {
-            key: params[key] for key in unassigned_params
-        }  # remove keys from params if they are assigned
+            # We have a parameter of this name, lets bind it
+            param_value = params[name]
+
+            if (
+                inspect.isclass(value.annotation)
+                and issubclass(value.annotation, BaseModel)
+            ) and not isinstance(param_value, ObjectParameter):
+                # Even if the annotation is a pydantic model, it can be passed as an object parameter
+                # We try to cast it as a pydantic model if asked
+                named_param = params[name].get_value()
+
+                if not isinstance(named_param, dict):
+                    # A case where the parameter is a one attribute model
+                    named_param = {name: named_param}
+
+                bound_model = bind_args_for_pydantic_model(
+                    named_param, value.annotation
+                )
+                bound_args[name] = bound_model
+
+            elif value.annotation in [str, int, float, bool] and callable(
+                value.annotation
+            ):
+                # Cast it if its a primitive type. Ensure the type matches the annotation.
+                try:
+                    bound_args[name] = value.annotation(params[name].get_value())
+                except (ValueError, TypeError) as e:
+                    raise ValueError(
+                        f"Cannot cast parameter '{name}' to {value.annotation.__name__}: {e}"
+                    )
+            else:
+                # We do not know type of parameter, we send the value as found
+                bound_args[name] = params[name].get_value()
+
+    # Find extra parameters (parameters in params but not consumed by regular function parameters)
+    consumed_param_names = set(bound_args.keys()) | set(missing_required_args)
+    extra_params = {k: v for k, v in params.items() if k not in consumed_param_names}
+
+    # Second pass: Handle **kwargs and argparse.Namespace parameters
+    if var_keyword_param is not None:
+        # Function accepts **kwargs - add all extra parameters directly to bound_args
+        for param_name, param_value in extra_params.items():
+            bound_args[param_name] = param_value.get_value()
+    elif namespace_param is not None:
+        # Function accepts argparse.Namespace - create namespace with extra parameters
+        args_namespace = argparse.Namespace()
+        for param_name, param_value in extra_params.items():
+            setattr(args_namespace, param_name, param_value.get_value())
+        bound_args[namespace_param] = args_namespace
+    elif extra_params:
+        # Function doesn't accept **kwargs or namespace, but we have extra parameters
+        # This should only be an error if we also have missing required parameters
+        # or if the function truly can't handle the extra parameters
+        if missing_required_args:
+            # We have both missing required and extra parameters - this is an error
+            raise ValueError(
+                f"Function {func.__name__} has parameters {missing_required_args} that are not present in the parameters"
+            )
+        # If we only have extra parameters and no missing required ones, we just ignore the extras
+        # This allows for more flexible parameter passing
+
+    # Check for missing required parameters
+    if missing_required_args:
+        if var_keyword_param is None and namespace_param is None:
+            # No way to handle missing parameters
+            raise ValueError(
+                f"Function {func.__name__} has parameters {missing_required_args} that are not present in the parameters"
+            )
+        # If we have **kwargs or namespace, missing parameters might be handled there
+        # But if they're truly required (no default), we should still error
+        raise ValueError(
+            f"Function {func.__name__} has parameters {missing_required_args} that are not present in the parameters"
+        )
 
     return bound_args