From 2b55d234da1738e8e2ea1f2776a9c97d8839a02f Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 11 Dec 2024 16:17:02 +0100 Subject: [PATCH 01/20] lightning example --- docs/infra/example_lightning.py | 148 ++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 docs/infra/example_lightning.py diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py new file mode 100644 index 00000000..9639f255 --- /dev/null +++ b/docs/infra/example_lightning.py @@ -0,0 +1,148 @@ + +from torchvision import datasets, transforms +from torchvision.models import resnet18 +import torch +import pydantic +import exca +import typing as tp +import sys +import pytorch_lightning as pl +from pytorch_lightning import Trainer +from pytorch_lightning.callbacks import ModelCheckpoint + + +class ResNet(pl.LightningModule): + def __init__(self, pretrained: bool=True, learning_rate: float=0.001): + super(ResNet, self).__init__() + self.pretrained = pretrained + self.learning_rate = learning_rate + self.model = resnet18(pretrained=pretrained) + self.model.conv1 = torch.nn.Conv2d(1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False) + self.model.fc = torch.nn.Linear(self.model.fc.in_features, 10) + self.loss_fn = torch.nn.CrossEntropyLoss() + + def forward(self, x): + return self.model(x) + + def _step(self, batch): + x, y = batch + y_hat = self(x) + loss = self.loss_fn(y_hat, y) + return loss + + def training_step(self, batch, batch_idx): + loss = self._step(batch) + self.log("train_loss", loss) + return loss + + def validation_step(self, batch, batch_idx): + loss = self._step(batch) + self.log("val_loss", loss) + return loss + + def configure_optimizers(self): + return torch.optim.Adam(self.model.parameters(), lr=self.learning_rate) + + +class Mnist(pl.LightningDataModule): + def __init__(self, batch_size=64): + super().__init__() + self.batch_size = batch_size + + def _dataloader(self, train: bool): + transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5,), (0.5,))]) + dset = datasets.MNIST('', train=train, download=True, transform=transform) + return torch.utils.data.DataLoader(dset, batch_size=self.batch_size) + + def train_dataloader(self): + return self._dataloader(train=True) + + def val_dataloader(self): + return self._dataloader(train=False) + + +class AutoConfig(pydantic.BaseModel): # TODO move to exca.helpers? + model_config = pydantic.ConfigDict(extra="forbid") + _cls: tp.ClassVar[tp.Any] + + def model_post_init(self, log__: tp.Any) -> None: + super().model_post_init(log__) + exca.helpers.validate_kwargs(self._cls, self._public_params()) + + def _public_params(self): + return {k: v for k, v in self.dict().items() if not k.startswith("_")} + + def build(self, **kwargs): + return self._cls(**self._public_params(), **kwargs) + +def args_to_nested_dict(args: list[str]) -> tp.Dict[str, tp.Any]: # TODO move to exca.helpers? + """ + Parses a list of Bash-style arguments (e.g., --key=value) into a nested dict. + """ + nested_dict = {} + for arg in args: + # Split argument into key and value + key, value = arg.lstrip("--").split("=", 1) + # Convert flat key into a nested dictionary + keys = key.split(".") + current_level = nested_dict + for k in keys[:-1]: + current_level = current_level.setdefault(k, {}) + current_level[keys[-1]] = value + return nested_dict + + +class ModelConfig(AutoConfig): # question: right design? + pretrained: bool = True + learning_rate: float = 0.001 + _cls = ResNet + + +class MnistConfig(AutoConfig): + batch_size: int = 64 # question: uid change if add new param, but corresponds to default trainer? + _cls = Mnist + + +class TrainerConfig(AutoConfig): + max_epochs: int = 5 + _cls = Trainer + + +class Experiment(pydantic.BaseModel): + model: ModelConfig = ModelConfig() + data: MnistConfig = MnistConfig() + trainer: TrainerConfig = TrainerConfig() + + infra: exca.TaskInfra = exca.TaskInfra() + + def build(self): + mnist = self.data.build() + model = self.model.build() + callbacks = None + if self.infra.folder: + callbacks = [ModelCheckpoint( + dirpath=self.infra.uid_folder() / 'checkpoint', + save_top_k=1, + monitor="val_loss", + mode="min") + ] # question: if training is preempted (how is the exca checkpoint loading going to behave?) + trainer = self.trainer.build(callbacks=callbacks) + return mnist, model, trainer + + @infra.apply + def fit(self): + data_loaders, model, trainer = self.build() + trainer.fit(model, data_loaders) + return model + + def validate(self): + data_loaders, _, trainer = self.build() + model = self.fit() + return trainer.validate(model, dataloaders=data_loaders.val_dataloader()) + + +if __name__ == '__main__': + config = args_to_nested_dict(sys.argv[1:]) + exp = Experiment(**config) + score = exp.validate() + print(score) \ No newline at end of file From 113e82804698580acf0fdf921f121d407b189c53 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 11 Dec 2024 16:23:24 +0100 Subject: [PATCH 02/20] up --- docs/infra/example_lightning.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index 9639f255..437ae1b5 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -145,4 +145,4 @@ def validate(self): config = args_to_nested_dict(sys.argv[1:]) exp = Experiment(**config) score = exp.validate() - print(score) \ No newline at end of file + print(score) From 5077a8bafa7505c3747dd363ef3b3b30a7aac85c Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 11 Dec 2024 16:50:36 +0100 Subject: [PATCH 03/20] add checkpoint --- docs/infra/example_lightning.py | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index 437ae1b5..900b8e83 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -1,14 +1,14 @@ +import sys +import typing as tp -from torchvision import datasets, transforms -from torchvision.models import resnet18 -import torch -import pydantic import exca -import typing as tp -import sys +import pydantic import pytorch_lightning as pl from pytorch_lightning import Trainer from pytorch_lightning.callbacks import ModelCheckpoint +import torch +from torchvision import datasets, transforms +from torchvision.models import resnet18 class ResNet(pl.LightningModule): @@ -125,14 +125,22 @@ def build(self): save_top_k=1, monitor="val_loss", mode="min") - ] # question: if training is preempted (how is the exca checkpoint loading going to behave?) + ] trainer = self.trainer.build(callbacks=callbacks) return mnist, model, trainer @infra.apply def fit(self): data_loaders, model, trainer = self.build() - trainer.fit(model, data_loaders) + # Define the checkpoint directory + checkpoint_dir = self.infra.uid_folder() / 'checkpoint' + + # Find the latest checkpoint if it exists + checkpoints = sorted(checkpoint_dir.glob('*.ckpt')) + ckpt_path = sorted(checkpoints)[-1] if checkpoints else None + + # Fit model + trainer.fit(model, data_loaders, ckpt_path=ckpt_path) return model def validate(self): From a23bccf9c9dfdb9eb7aacf9d4bfa287437ac71a5 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 11 Dec 2024 17:29:31 +0100 Subject: [PATCH 04/20] check that default config match original class --- docs/infra/example_lightning.py | 50 ++++++++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 7 deletions(-) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index 900b8e83..d516a952 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -1,3 +1,4 @@ +import inspect import sys import typing as tp @@ -65,15 +66,50 @@ class AutoConfig(pydantic.BaseModel): # TODO move to exca.helpers? model_config = pydantic.ConfigDict(extra="forbid") _cls: tp.ClassVar[tp.Any] + @classmethod + def __pydantic_init_subclass__(cls, **kwargs: tp.Any) -> None: + """Checks that the config default values match the _cls defaults""" + super().__pydantic_init_subclass__(**kwargs) + super().__init_subclass__() + + if isinstance(cls._cls, type): + func_or_class = cls._cls.__init__ + else: + func_or_class = cls._cls + + # Get the function signature + signature = inspect.signature(func_or_class) + func_params = signature.parameters + + # Iterate through the class fields and verify their defaults + for field_name, field_info in cls.model_fields.items(): + # Check if the field has a default value or is required + model_default = field_info.default + model_required = field_info.is_required() + + # Check if the parameter exists in the function signature + if field_name not in func_params: + raise ValueError(f"Field '{field_name}' is missing in the function parameters.") + + func_param = func_params[field_name] + func_default = func_param.default + + # Check if the field is required in both the function and the model + if model_required != (func_default is inspect.Parameter.empty): + raise ValueError(f"Field '{field_name}' is required in the model but not in the function or vice versa.") + + # If it has a default in both, compare them + if model_default != func_default and func_default is not inspect.Parameter.empty: + raise ValueError(f"Field '{field_name}' default value mismatch: model has '{model_default}', function has '{func_default}'.") + def model_post_init(self, log__: tp.Any) -> None: + """Check that the parameters are compatible with _cls""" super().model_post_init(log__) - exca.helpers.validate_kwargs(self._cls, self._public_params()) + exca.helpers.validate_kwargs(self._cls, self.dict()) - def _public_params(self): - return {k: v for k, v in self.dict().items() if not k.startswith("_")} + def build(self, **kwargs): # /!\ **kwargs needed for trainer checkpoint, but bad api for uid? + return self._cls(**self.dict(), **kwargs) - def build(self, **kwargs): - return self._cls(**self._public_params(), **kwargs) def args_to_nested_dict(args: list[str]) -> tp.Dict[str, tp.Any]: # TODO move to exca.helpers? """ @@ -104,7 +140,7 @@ class MnistConfig(AutoConfig): class TrainerConfig(AutoConfig): - max_epochs: int = 5 + max_epochs: tp.Optional[int] = None _cls = Trainer @@ -150,7 +186,7 @@ def validate(self): if __name__ == '__main__': - config = args_to_nested_dict(sys.argv[1:]) + config = args_to_nested_dict(['--trainer.max_epochs=5'] + sys.argv[1:]) exp = Experiment(**config) score = exp.validate() print(score) From 10ae19b9cec93421c3ea849250a53fbfac03f534 Mon Sep 17 00:00:00 2001 From: kingjr Date: Fri, 13 Dec 2024 21:34:34 +0100 Subject: [PATCH 05/20] address comments --- docs/infra/example_lightning.py | 170 +++++++++++--------------------- 1 file changed, 59 insertions(+), 111 deletions(-) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index d516a952..5717dc20 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -1,5 +1,3 @@ -import inspect -import sys import typing as tp import exca @@ -12,12 +10,27 @@ from torchvision.models import resnet18 -class ResNet(pl.LightningModule): - def __init__(self, pretrained: bool=True, learning_rate: float=0.001): - super(ResNet, self).__init__() - self.pretrained = pretrained - self.learning_rate = learning_rate - self.model = resnet18(pretrained=pretrained) +class ModelConfig(pydantic.BaseModel): + pretrained: bool = True + learning_rate: float = 0.001 + model_config = pydantic.ConfigDict(extra="forbid") + + +class DataConfig(pydantic.BaseModel): + batch_size: int = 64 + model_config = pydantic.ConfigDict(extra="forbid") + + +class TrainerConfig(pydantic.BaseModel): + max_epochs: tp.Optional[int] = None + model_config = pydantic.ConfigDict(extra="forbid") + + +class Model(pl.LightningModule): + def __init__(self, config: ModelConfig = ModelConfig()): + super(Model, self).__init__() + self.config = config + self.model = resnet18(pretrained=config.pretrained) self.model.conv1 = torch.nn.Conv2d(1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False) self.model.fc = torch.nn.Linear(self.model.fc.in_features, 10) self.loss_fn = torch.nn.CrossEntropyLoss() @@ -42,118 +55,37 @@ def validation_step(self, batch, batch_idx): return loss def configure_optimizers(self): - return torch.optim.Adam(self.model.parameters(), lr=self.learning_rate) + return torch.optim.Adam(self.model.parameters(), lr=self.config.learning_rate) -class Mnist(pl.LightningDataModule): - def __init__(self, batch_size=64): +class Data(pl.LightningDataModule): + def __init__(self, config: DataConfig): super().__init__() - self.batch_size = batch_size + self.config = config def _dataloader(self, train: bool): transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5,), (0.5,))]) dset = datasets.MNIST('', train=train, download=True, transform=transform) - return torch.utils.data.DataLoader(dset, batch_size=self.batch_size) + return torch.utils.data.DataLoader(dset, batch_size=self.config.batch_size) def train_dataloader(self): return self._dataloader(train=True) def val_dataloader(self): return self._dataloader(train=False) - - -class AutoConfig(pydantic.BaseModel): # TODO move to exca.helpers? - model_config = pydantic.ConfigDict(extra="forbid") - _cls: tp.ClassVar[tp.Any] - - @classmethod - def __pydantic_init_subclass__(cls, **kwargs: tp.Any) -> None: - """Checks that the config default values match the _cls defaults""" - super().__pydantic_init_subclass__(**kwargs) - super().__init_subclass__() - - if isinstance(cls._cls, type): - func_or_class = cls._cls.__init__ - else: - func_or_class = cls._cls - - # Get the function signature - signature = inspect.signature(func_or_class) - func_params = signature.parameters - - # Iterate through the class fields and verify their defaults - for field_name, field_info in cls.model_fields.items(): - # Check if the field has a default value or is required - model_default = field_info.default - model_required = field_info.is_required() - - # Check if the parameter exists in the function signature - if field_name not in func_params: - raise ValueError(f"Field '{field_name}' is missing in the function parameters.") - - func_param = func_params[field_name] - func_default = func_param.default - - # Check if the field is required in both the function and the model - if model_required != (func_default is inspect.Parameter.empty): - raise ValueError(f"Field '{field_name}' is required in the model but not in the function or vice versa.") - - # If it has a default in both, compare them - if model_default != func_default and func_default is not inspect.Parameter.empty: - raise ValueError(f"Field '{field_name}' default value mismatch: model has '{model_default}', function has '{func_default}'.") - - def model_post_init(self, log__: tp.Any) -> None: - """Check that the parameters are compatible with _cls""" - super().model_post_init(log__) - exca.helpers.validate_kwargs(self._cls, self.dict()) - def build(self, **kwargs): # /!\ **kwargs needed for trainer checkpoint, but bad api for uid? - return self._cls(**self.dict(), **kwargs) - - -def args_to_nested_dict(args: list[str]) -> tp.Dict[str, tp.Any]: # TODO move to exca.helpers? - """ - Parses a list of Bash-style arguments (e.g., --key=value) into a nested dict. - """ - nested_dict = {} - for arg in args: - # Split argument into key and value - key, value = arg.lstrip("--").split("=", 1) - # Convert flat key into a nested dictionary - keys = key.split(".") - current_level = nested_dict - for k in keys[:-1]: - current_level = current_level.setdefault(k, {}) - current_level[keys[-1]] = value - return nested_dict - - -class ModelConfig(AutoConfig): # question: right design? - pretrained: bool = True - learning_rate: float = 0.001 - _cls = ResNet - - -class MnistConfig(AutoConfig): - batch_size: int = 64 # question: uid change if add new param, but corresponds to default trainer? - _cls = Mnist - - -class TrainerConfig(AutoConfig): - max_epochs: tp.Optional[int] = None - _cls = Trainer - class Experiment(pydantic.BaseModel): model: ModelConfig = ModelConfig() - data: MnistConfig = MnistConfig() + data: DataConfig = DataConfig() trainer: TrainerConfig = TrainerConfig() - - infra: exca.TaskInfra = exca.TaskInfra() + infra: exca.TaskInfra = exca.TaskInfra(folder='.cache/') def build(self): - mnist = self.data.build() - model = self.model.build() + data = Data(self.data) + model = Model(self.model) + + # Add checkpoint callback based on infra folder callbacks = None if self.infra.folder: callbacks = [ModelCheckpoint( @@ -162,31 +94,47 @@ def build(self): monitor="val_loss", mode="min") ] - trainer = self.trainer.build(callbacks=callbacks) - return mnist, model, trainer - @infra.apply - def fit(self): - data_loaders, model, trainer = self.build() + trainer = Trainer(**self.trainer.dict(), callbacks=callbacks) + return data, model, trainer + + @property + def checkpoint_path(self): # Define the checkpoint directory checkpoint_dir = self.infra.uid_folder() / 'checkpoint' # Find the latest checkpoint if it exists checkpoints = sorted(checkpoint_dir.glob('*.ckpt')) ckpt_path = sorted(checkpoints)[-1] if checkpoints else None + return ckpt_path + + @infra.apply + def fit(self): + # Configure + data_loaders, model, trainer = self.build() # Fit model - trainer.fit(model, data_loaders, ckpt_path=ckpt_path) - return model + trainer.fit(model, data_loaders, ckpt_path=self.checkpoint_path) + + # Return model if not saved + if self.checkpoint_path is None: + return model def validate(self): - data_loaders, _, trainer = self.build() - model = self.fit() - return trainer.validate(model, dataloaders=data_loaders.val_dataloader()) + data_loaders, model, trainer = self.build() + trained_model = self.fit() + if trained_model is None: + trained_model = model.__class__.load_from_checkpoint(self.checkpoint_path) + + return trainer.validate(trained_model, dataloaders=data_loaders.val_dataloader()) if __name__ == '__main__': - config = args_to_nested_dict(['--trainer.max_epochs=5'] + sys.argv[1:]) + config = dict( + model={'learning_rate': .01}, + trainer={'max_epochs': 2}, + infra={'folder': '.cache/'} + ) exp = Experiment(**config) score = exp.validate() print(score) From e1d381ae310ce9a3a5a7d29c55d9ee6b813c4f26 Mon Sep 17 00:00:00 2001 From: kingjr Date: Mon, 16 Dec 2024 13:04:34 +0100 Subject: [PATCH 06/20] config.build --- docs/infra/example_lightning.py | 96 ++++++++++++++++++--------------- 1 file changed, 53 insertions(+), 43 deletions(-) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index 5717dc20..b274a0ca 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -10,27 +10,12 @@ from torchvision.models import resnet18 -class ModelConfig(pydantic.BaseModel): - pretrained: bool = True - learning_rate: float = 0.001 - model_config = pydantic.ConfigDict(extra="forbid") - - -class DataConfig(pydantic.BaseModel): - batch_size: int = 64 - model_config = pydantic.ConfigDict(extra="forbid") - - -class TrainerConfig(pydantic.BaseModel): - max_epochs: tp.Optional[int] = None - model_config = pydantic.ConfigDict(extra="forbid") - - class Model(pl.LightningModule): - def __init__(self, config: ModelConfig = ModelConfig()): + def __init__(self, pretrained: bool, learning_rate: float = 0.001): super(Model, self).__init__() - self.config = config - self.model = resnet18(pretrained=config.pretrained) + self.pretrained = pretrained + self.learning_rate = learning_rate + self.model = resnet18(pretrained=pretrained) self.model.conv1 = torch.nn.Conv2d(1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False) self.model.fc = torch.nn.Linear(self.model.fc.in_features, 10) self.loss_fn = torch.nn.CrossEntropyLoss() @@ -55,18 +40,18 @@ def validation_step(self, batch, batch_idx): return loss def configure_optimizers(self): - return torch.optim.Adam(self.model.parameters(), lr=self.config.learning_rate) + return torch.optim.Adam(self.model.parameters(), lr=self.learning_rate) class Data(pl.LightningDataModule): - def __init__(self, config: DataConfig): + def __init__(self, batch_size: int): super().__init__() - self.config = config + self.batch_size = batch_size def _dataloader(self, train: bool): transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5,), (0.5,))]) dset = datasets.MNIST('', train=train, download=True, transform=transform) - return torch.utils.data.DataLoader(dset, batch_size=self.config.batch_size) + return torch.utils.data.DataLoader(dset, batch_size=self.batch_size) def train_dataloader(self): return self._dataloader(train=True) @@ -75,28 +60,48 @@ def val_dataloader(self): return self._dataloader(train=False) -class Experiment(pydantic.BaseModel): - model: ModelConfig = ModelConfig() - data: DataConfig = DataConfig() - trainer: TrainerConfig = TrainerConfig() - infra: exca.TaskInfra = exca.TaskInfra(folder='.cache/') - def build(self): - data = Data(self.data) - model = Model(self.model) - - # Add checkpoint callback based on infra folder - callbacks = None - if self.infra.folder: +class ModelConfig(pydantic.BaseModel): + pretrained: bool = True + learning_rate: float = 0.001 + + model_config = pydantic.ConfigDict(extra="forbid") + + def build(self) -> Model: + return Model(**self.dict()) + + +class DataConfig(pydantic.BaseModel): + batch_size: int = 64 + + model_config = pydantic.ConfigDict(extra="forbid") + + def build(self) -> Data: + return Data(**self.dict()) + + +class TrainerConfig(pydantic.BaseModel): + max_epochs: tp.Optional[int] = None + + model_config = pydantic.ConfigDict(extra="forbid") + + def build(self, checkpoint_path: str | None = None) -> Trainer: + if checkpoint_path: callbacks = [ModelCheckpoint( - dirpath=self.infra.uid_folder() / 'checkpoint', + dirpath=checkpoint_path, save_top_k=1, monitor="val_loss", mode="min") ] + else: + callbacks = None + return Trainer(**self.dict(), callbacks=callbacks) - trainer = Trainer(**self.trainer.dict(), callbacks=callbacks) - return data, model, trainer +class Experiment(pydantic.BaseModel): + model: ModelConfig = ModelConfig() + data: DataConfig = DataConfig() + trainer: TrainerConfig = TrainerConfig() + infra: exca.TaskInfra = exca.TaskInfra(folder='.cache/') @property def checkpoint_path(self): @@ -111,22 +116,27 @@ def checkpoint_path(self): @infra.apply def fit(self): # Configure - data_loaders, model, trainer = self.build() - + data = self.data.build() + model = self.model.build() + trainer = self.trainer.build(self.infra.folder) + # Fit model - trainer.fit(model, data_loaders, ckpt_path=self.checkpoint_path) + trainer.fit(model, data, ckpt_path=self.checkpoint_path) # Return model if not saved if self.checkpoint_path is None: return model def validate(self): - data_loaders, model, trainer = self.build() + data = self.data.build() + model = self.model.build() + trainer = self.trainer.build(self.infra.folder) + trained_model = self.fit() if trained_model is None: trained_model = model.__class__.load_from_checkpoint(self.checkpoint_path) - return trainer.validate(trained_model, dataloaders=data_loaders.val_dataloader()) + return trainer.validate(trained_model, dataloaders=data.val_dataloader()) if __name__ == '__main__': From 9c36d9c4c21897abd3cc3958d52ec4d7b9aeb871 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=A9my=20Rapin?= Date: Tue, 24 Dec 2024 10:45:58 +0100 Subject: [PATCH 07/20] Update example_lightning.py --- docs/infra/example_lightning.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/infra/example_lightning.py b/docs/infra/example_lightning.py index b274a0ca..08426c2a 100644 --- a/docs/infra/example_lightning.py +++ b/docs/infra/example_lightning.py @@ -1,3 +1,8 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the license found in the +# LICENSE file in the root directory of this source tree. import typing as tp import exca From 7912ba6ff61e214a26ce09c10e4e433a93cb0702 Mon Sep 17 00:00:00 2001 From: Jeremy Rapin Date: Tue, 24 Dec 2024 11:28:18 +0100 Subject: [PATCH 08/20] Add packages for examples in docs --- .github/workflows/test-type-lint.yaml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/test-type-lint.yaml b/.github/workflows/test-type-lint.yaml index 51e4241f..629d13c0 100644 --- a/.github/workflows/test-type-lint.yaml +++ b/.github/workflows/test-type-lint.yaml @@ -48,6 +48,7 @@ jobs: run: | source activate ./ci_env pip install -e .[dev] + pip install sklearn lightning # for docs - name: Print installed packages run: | @@ -76,7 +77,7 @@ jobs: sed -i 's/\"auto\"/None/g' README.md # on Mac: sed -i '' 's/cluster: slurm/cluster: null/g' infra/*.md # check readmes - pytest --markdown-docs -m markdown-docs `**/*.md` + pytest --markdown-docs -m markdown-docs . - name: Run basic pylint run: | From 642aeb05e0ff8576c1d4cc5b15b907d1bc565069 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=A9my=20Rapin?= Date: Tue, 24 Dec 2024 11:48:44 +0100 Subject: [PATCH 09/20] Update .github/workflows/test-type-lint.yaml --- .github/workflows/test-type-lint.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test-type-lint.yaml b/.github/workflows/test-type-lint.yaml index 629d13c0..cc1ee818 100644 --- a/.github/workflows/test-type-lint.yaml +++ b/.github/workflows/test-type-lint.yaml @@ -48,7 +48,7 @@ jobs: run: | source activate ./ci_env pip install -e .[dev] - pip install sklearn lightning # for docs + pip install scikit-learn lightning # for docs - name: Print installed packages run: | From 588ef196ca5034019aada600820a40f88caa742c Mon Sep 17 00:00:00 2001 From: Jeremy Rapin Date: Tue, 24 Dec 2024 12:43:54 +0100 Subject: [PATCH 10/20] add --- .github/workflows/test-type-lint.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test-type-lint.yaml b/.github/workflows/test-type-lint.yaml index cc1ee818..b6122f3c 100644 --- a/.github/workflows/test-type-lint.yaml +++ b/.github/workflows/test-type-lint.yaml @@ -48,7 +48,7 @@ jobs: run: | source activate ./ci_env pip install -e .[dev] - pip install scikit-learn lightning # for docs + pip install scikit-learn lightning torchvision # for docs - name: Print installed packages run: | From bc98b748873a7f6b3dac70018f542fb5396b6c6e Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 16:45:21 +0100 Subject: [PATCH 11/20] Add to_step and to_chain helpers for function-to-Step conversion - to_step(func): creates a Step subclass from a plain function, auto-detecting pipeline inputs (params without defaults) - to_chain(*funcs): creates a parameterizable Chain subclass with per-function fields; supports (name, func) tuples for disambiguation Co-authored-by: Cursor --- exca/steps/__init__.py | 2 + exca/steps/base.py | 207 +++++++++++++++++++++++++++++++++++++++ exca/steps/test_steps.py | 97 +++++++++++++++++- 3 files changed, 305 insertions(+), 1 deletion(-) diff --git a/exca/steps/__init__.py b/exca/steps/__init__.py index 747c7934..d027171d 100644 --- a/exca/steps/__init__.py +++ b/exca/steps/__init__.py @@ -28,3 +28,5 @@ from . import backends from .base import Chain as Chain from .base import Step as Step +from .base import to_chain as to_chain +from .base import to_step as to_step diff --git a/exca/steps/base.py b/exca/steps/base.py index 00b304a0..8402fe0f 100644 --- a/exca/steps/base.py +++ b/exca/steps/base.py @@ -25,6 +25,8 @@ import exca from exca import utils +X = tp.TypeVar("X") + from . import backends from .backends import NoValue @@ -425,3 +427,208 @@ def clear_cache(self, recursive: bool = True) -> None: step.clear_cache() if self.infra: self.infra.clear_cache() + + +def _resolve_input_params( + func: tp.Callable[..., tp.Any], + input_params: tp.Sequence[str] | None, +) -> tuple[str, ...]: + """Determine which function parameters are pipeline inputs. + + If *input_params* is ``None``, parameters without a default value are + used. Returns a validated tuple of parameter names. + """ + sig = inspect.signature(func) + if input_params is None: + return tuple( + p.name + for p in sig.parameters.values() + if p.default is inspect._empty + ) + for name in input_params: + if name not in sig.parameters: + raise ValueError( + f"input_params contains {name!r} which is not a parameter " + f"of {func.__name__}; available: {list(sig.parameters)}" + ) + return tuple(input_params) + + +def to_step( + func: tp.Callable[..., X], + input_params: tp.Sequence[str] | None = None, +) -> tp.Type[Step]: + """Create a Step subclass from a function. + + Parameters + ---------- + func: + The function to wrap. Parameters that are **not** pipeline inputs + must have type annotations and become pydantic model fields. + input_params: + Names of parameters that receive pipeline input at runtime. + By default (``None``), every parameter **without a default value** + is treated as a pipeline input. Pass an explicit list (possibly + empty) to override. + + - 0 input params -> generator step, ``_run(self)`` + - 1 input param -> ``_run(self, value)`` + - N input params -> ``_run(self, value)`` where *value* is + unpacked as a tuple of length N + + Example + ------- + >>> def multiply(value: float, coeff: float = 2.0) -> float: + ... return value * coeff + >>> MultiplyStep = to_step(multiply) # value has no default -> input + >>> MultiplyStep(coeff=3).run(5.0) + 15.0 + """ + resolved = _resolve_input_params(func, input_params) + resolved_set = set(resolved) + reserved = {"infra"} + sig = inspect.signature(func) + + fields: dict[str, tp.Any] = {} + for p in sig.parameters.values(): + if p.name in resolved_set: + continue + if p.name in reserved: + raise ValueError( + f"Parameter {p.name!r} is reserved and cannot be used " + "as a field name" + ) + if p.annotation in (tp.Any, inspect._empty): + raise ValueError( + f"Parameter {p.name!r} of {func.__name__} needs a type " + "annotation" + ) + default = ... if p.default is inspect._empty else p.default + fields[p.name] = (p.annotation, default) + + # Build _run that calls the original function + if len(resolved) == 0: + def _run(self: tp.Any) -> tp.Any: # type: ignore[misc] + return func(**{n: getattr(self, n) for n in fields}) + elif len(resolved) == 1: + _input_name = resolved[0] + + def _run(self: tp.Any, value: tp.Any) -> tp.Any: # type: ignore[misc] + kwargs = {n: getattr(self, n) for n in fields} + kwargs[_input_name] = value + return func(**kwargs) + else: + _input_names = resolved + + def _run(self: tp.Any, value: tp.Any) -> tp.Any: # type: ignore[misc] + kwargs = {n: getattr(self, n) for n in fields} + kwargs.update(zip(_input_names, value)) + return func(**kwargs) + + Model: tp.Type[Step] = pydantic.create_model( + func.__name__ + "_Step", + **fields, + __base__=Step, + __module__=func.__module__, + ) + Model._run = _run # type: ignore[assignment] + return Model + + +_FuncOrNamed = tp.Union[ + tp.Callable[..., tp.Any], + tp.Tuple[str, tp.Callable[..., tp.Any]], +] + + +def to_chain( + *funcs: _FuncOrNamed, + infra: tp.Any = None, +) -> tp.Type[Chain]: + """Create a Chain subclass from plain functions. + + Each function is converted to a Step via :func:`to_step` and exposed + as a field on the returned Chain subclass, keyed by the function + name. This lets you parameterize each step independently. + + To use the same function more than once, pass ``(name, func)`` + tuples to give each occurrence a distinct field name. + + Parameters + ---------- + *funcs: + Functions (or ``(name, func)`` tuples) to chain sequentially. + infra: + Optional default infra for the Chain (e.g. + ``{"backend": "Cached", "folder": "/tmp/cache"}``). + + Example + ------- + >>> def generate(seed: int = 42) -> float: + ... import random; return random.Random(seed).random() + >>> def scale(x: float, factor: float = 10.0) -> float: + ... return x * factor + >>> MyChain = to_chain(generate, ("upscale", scale), ("downscale", scale)) + >>> chain = MyChain(upscale=dict(factor=100), downscale=dict(factor=0.5)) + >>> result = chain.run() + """ + if not funcs: + raise ValueError("to_chain requires at least one function") + + # Normalise inputs to (name, func) pairs + named: list[tuple[str, tp.Callable[..., tp.Any]]] = [] + for entry in funcs: + if isinstance(entry, tuple): + name, func = entry + named.append((name, func)) + else: + named.append((entry.__name__, entry)) + + # Check for duplicate field names + seen: dict[str, int] = {} + for i, (name, _) in enumerate(named): + if name in seen: + first = named[seen[name]][1].__name__ + current = named[i][1].__name__ + raise ValueError( + f"Duplicate field name {name!r} (from {first!r} and " + f"{current!r}); use (name, func) tuples to disambiguate" + ) + seen[name] = i + + step_classes = [to_step(f) for _, f in named] + _field_names = tuple(name for name, _ in named) + _default_infra = infra + + # One field per step, typed as its Step subclass (with a default + # instance so all fields are optional). + model_fields: dict[str, tp.Any] = {} + for (name, _), StepCls in zip(named, step_classes): + model_fields[name] = (StepCls, StepCls()) + + # Override ``steps`` with an empty default; it is rebuilt from the + # per-function fields in model_post_init. + model_fields["steps"] = (tp.Sequence[Step], ()) + + chain_name = "_".join(_field_names) + "_Chain" + + Model: tp.Type[Chain] = pydantic.create_model( + chain_name, + **model_fields, + __base__=Chain, + __module__=named[0][1].__module__, + ) + + _super_post_init = Chain.model_post_init + + def _model_post_init(self: tp.Any, __context: tp.Any) -> None: + if not self.steps: + built = tuple(getattr(self, name) for name in _field_names) + object.__setattr__(self, "steps", built) + if _default_infra is not None and self.infra is None: + object.__setattr__(self, "infra", _default_infra) + _super_post_init(self, __context) + + Model.model_post_init = _model_post_init # type: ignore[assignment] + + return Model diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index c319b801..0f6341a1 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -16,7 +16,7 @@ import exca from . import backends, conftest -from .base import Chain, Input, Step +from .base import Chain, Input, Step, to_chain, to_step # ============================================================================= # Basic execution (no infra) @@ -312,3 +312,98 @@ def _forward(self, x: float) -> float: # .forward() call triggers its own warning with pytest.warns(DeprecationWarning, match="forward.*deprecated.*run"): assert step.forward(5.0) == 10.0 + + +# ============================================================================= +# to_step / to_chain helpers +# ============================================================================= + +import random as _random + + +def _generate(seed: int = 42) -> float: + return _random.Random(seed).random() + + +def _scale(x: float, factor: float = 10.0) -> float: + return x * factor + + +def test_to_step() -> None: + # generator (all defaults) + G = to_step(_generate) + g = G(seed=123) + assert g._is_generator() + assert g.run() == g.run() + # transformer (auto-detect required param as input) + M = to_step(_scale) + assert not M(coeff=3.0).run(5.0) != 15.0 # noqa (negated for conciseness) + assert M().run(5.0) == 50.0 + # explicit input_params override + def add(a: float, b: float) -> float: + return a + b + assert to_step(add, input_params=["a"])(b=10.0).run(5.0) == 15.0 + # multiple inputs -> tuple unpacking + def combine(x: int, y: int, s: float = 1.0) -> float: + return (x + y) * s + assert to_step(combine)(s=2.0).run((3, 7)) == 20.0 + # in a Chain + chain = Chain(steps=[G(seed=42), M(factor=100.0)]) + assert chain.run() == pytest.approx(_random.Random(42).random() * 100.0) + + +def test_to_step_with_infra(tmp_path: Path) -> None: + G = to_step(_generate) + infra: tp.Any = {"backend": "Cached", "folder": tmp_path} + step = G(seed=7, infra=infra) + assert step.run() == step.run() + assert step.with_input().has_cache() + + +def test_to_step_validation_errors() -> None: + def f(x: int) -> int: + return x + with pytest.raises(ValueError, match="not a parameter"): + to_step(f, input_params=["nope"]) + def g(x, y: int = 1) -> int: # type: ignore[no-untyped-def] + return x + y + with pytest.raises(ValueError, match="needs a type annotation"): + to_step(g, input_params=["x"]) + def h(infra: int) -> int: + return infra + with pytest.raises(ValueError, match="reserved"): + to_step(h, input_params=[]) + + +def test_to_chain() -> None: + MyChain = to_chain(_generate, _scale) + assert issubclass(MyChain, Chain) + # defaults + assert MyChain().run() == pytest.approx(_random.Random(42).random() * 10.0) + # custom + c = MyChain(generate=dict(seed=123), scale=dict(factor=100.0)) + assert c.run() == pytest.approx(_random.Random(123).random() * 100.0) + # partial + c2 = MyChain(scale=dict(factor=5.0)) + assert c2.run() == pytest.approx(_random.Random(42).random() * 5.0) + + +def test_to_chain_with_infra(tmp_path: Path) -> None: + def double(x: float) -> float: + return x * 2 + MyChain = to_chain(_generate, double, infra={"backend": "Cached", "folder": tmp_path}) + chain = MyChain() + assert chain.run() == chain.run() + + +def test_to_chain_named_and_errors() -> None: + # (name, func) tuples for duplicate functions + MyChain = to_chain(_generate, ("up", _scale), ("down", _scale)) + c = MyChain(up=dict(factor=100.0), down=dict(factor=0.5)) + assert c.run() == pytest.approx(_generate() * 100.0 * 0.5) + # duplicate bare names rejected + with pytest.raises(ValueError, match="Duplicate"): + to_chain(_scale, _scale) + # empty rejected + with pytest.raises(ValueError, match="at least one"): + to_chain() From da885c15a83d1f068439b98e29a68cda1d61d28d Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 16:50:05 +0100 Subject: [PATCH 12/20] Fix mypy errors: add type: ignore for dynamic model fields pydantic.create_model returns type[BaseModel] so mypy cannot see the dynamically-created fields from to_step/to_chain. Co-authored-by: Cursor --- exca/steps/test_steps.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index 0f6341a1..feb406bc 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -332,30 +332,30 @@ def _scale(x: float, factor: float = 10.0) -> float: def test_to_step() -> None: # generator (all defaults) G = to_step(_generate) - g = G(seed=123) + g = G(seed=123) # type: ignore[call-arg] assert g._is_generator() assert g.run() == g.run() # transformer (auto-detect required param as input) M = to_step(_scale) - assert not M(coeff=3.0).run(5.0) != 15.0 # noqa (negated for conciseness) + assert M(coeff=3.0).run(5.0) == 15.0 # type: ignore[call-arg] assert M().run(5.0) == 50.0 # explicit input_params override def add(a: float, b: float) -> float: return a + b - assert to_step(add, input_params=["a"])(b=10.0).run(5.0) == 15.0 + assert to_step(add, input_params=["a"])(b=10.0).run(5.0) == 15.0 # type: ignore[call-arg] # multiple inputs -> tuple unpacking def combine(x: int, y: int, s: float = 1.0) -> float: return (x + y) * s - assert to_step(combine)(s=2.0).run((3, 7)) == 20.0 + assert to_step(combine)(s=2.0).run((3, 7)) == 20.0 # type: ignore[call-arg] # in a Chain - chain = Chain(steps=[G(seed=42), M(factor=100.0)]) + chain = Chain(steps=[G(seed=42), M(factor=100.0)]) # type: ignore[call-arg] assert chain.run() == pytest.approx(_random.Random(42).random() * 100.0) def test_to_step_with_infra(tmp_path: Path) -> None: G = to_step(_generate) infra: tp.Any = {"backend": "Cached", "folder": tmp_path} - step = G(seed=7, infra=infra) + step = G(seed=7, infra=infra) # type: ignore[call-arg] assert step.run() == step.run() assert step.with_input().has_cache() @@ -379,12 +379,12 @@ def test_to_chain() -> None: MyChain = to_chain(_generate, _scale) assert issubclass(MyChain, Chain) # defaults - assert MyChain().run() == pytest.approx(_random.Random(42).random() * 10.0) + assert MyChain().run() == pytest.approx(_random.Random(42).random() * 10.0) # type: ignore[call-arg] # custom - c = MyChain(generate=dict(seed=123), scale=dict(factor=100.0)) + c = MyChain(generate=dict(seed=123), scale=dict(factor=100.0)) # type: ignore[call-arg] assert c.run() == pytest.approx(_random.Random(123).random() * 100.0) # partial - c2 = MyChain(scale=dict(factor=5.0)) + c2 = MyChain(scale=dict(factor=5.0)) # type: ignore[call-arg] assert c2.run() == pytest.approx(_random.Random(42).random() * 5.0) @@ -392,14 +392,14 @@ def test_to_chain_with_infra(tmp_path: Path) -> None: def double(x: float) -> float: return x * 2 MyChain = to_chain(_generate, double, infra={"backend": "Cached", "folder": tmp_path}) - chain = MyChain() + chain = MyChain() # type: ignore[call-arg] assert chain.run() == chain.run() def test_to_chain_named_and_errors() -> None: # (name, func) tuples for duplicate functions MyChain = to_chain(_generate, ("up", _scale), ("down", _scale)) - c = MyChain(up=dict(factor=100.0), down=dict(factor=0.5)) + c = MyChain(up=dict(factor=100.0), down=dict(factor=0.5)) # type: ignore[call-arg] assert c.run() == pytest.approx(_generate() * 100.0 * 0.5) # duplicate bare names rejected with pytest.raises(ValueError, match="Duplicate"): From 747ff86daed5332c64f2304f9089d791905d5093 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 16:55:51 +0100 Subject: [PATCH 13/20] Fix test failures: remove leading underscores from helpers, fix typos - Rename _generate/_scale to generate/scale (pydantic treats underscore-prefixed names as private attrs in create_model) - Fix M(coeff=3.0) -> M(factor=3.0) to match _scale's param name - Fix validation test: unannotated param must be a non-input field Co-authored-by: Cursor --- exca/steps/test_steps.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index feb406bc..ceaef4aa 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -321,23 +321,23 @@ def _forward(self, x: float) -> float: import random as _random -def _generate(seed: int = 42) -> float: +def generate(seed: int = 42) -> float: return _random.Random(seed).random() -def _scale(x: float, factor: float = 10.0) -> float: +def scale(x: float, factor: float = 10.0) -> float: return x * factor def test_to_step() -> None: # generator (all defaults) - G = to_step(_generate) + G = to_step(generate) g = G(seed=123) # type: ignore[call-arg] assert g._is_generator() assert g.run() == g.run() # transformer (auto-detect required param as input) - M = to_step(_scale) - assert M(coeff=3.0).run(5.0) == 15.0 # type: ignore[call-arg] + M = to_step(scale) + assert M(factor=3.0).run(5.0) == 15.0 # type: ignore[call-arg] assert M().run(5.0) == 50.0 # explicit input_params override def add(a: float, b: float) -> float: @@ -353,7 +353,7 @@ def combine(x: int, y: int, s: float = 1.0) -> float: def test_to_step_with_infra(tmp_path: Path) -> None: - G = to_step(_generate) + G = to_step(generate) infra: tp.Any = {"backend": "Cached", "folder": tmp_path} step = G(seed=7, infra=infra) # type: ignore[call-arg] assert step.run() == step.run() @@ -365,7 +365,7 @@ def f(x: int) -> int: return x with pytest.raises(ValueError, match="not a parameter"): to_step(f, input_params=["nope"]) - def g(x, y: int = 1) -> int: # type: ignore[no-untyped-def] + def g(x: int, y) -> int: # type: ignore[no-untyped-def] return x + y with pytest.raises(ValueError, match="needs a type annotation"): to_step(g, input_params=["x"]) @@ -376,7 +376,7 @@ def h(infra: int) -> int: def test_to_chain() -> None: - MyChain = to_chain(_generate, _scale) + MyChain = to_chain(generate, scale) assert issubclass(MyChain, Chain) # defaults assert MyChain().run() == pytest.approx(_random.Random(42).random() * 10.0) # type: ignore[call-arg] @@ -391,19 +391,19 @@ def test_to_chain() -> None: def test_to_chain_with_infra(tmp_path: Path) -> None: def double(x: float) -> float: return x * 2 - MyChain = to_chain(_generate, double, infra={"backend": "Cached", "folder": tmp_path}) + MyChain = to_chain(generate, double, infra={"backend": "Cached", "folder": tmp_path}) chain = MyChain() # type: ignore[call-arg] assert chain.run() == chain.run() def test_to_chain_named_and_errors() -> None: # (name, func) tuples for duplicate functions - MyChain = to_chain(_generate, ("up", _scale), ("down", _scale)) + MyChain = to_chain(generate, ("up", scale), ("down", scale)) c = MyChain(up=dict(factor=100.0), down=dict(factor=0.5)) # type: ignore[call-arg] - assert c.run() == pytest.approx(_generate() * 100.0 * 0.5) + assert c.run() == pytest.approx(generate() * 100.0 * 0.5) # duplicate bare names rejected with pytest.raises(ValueError, match="Duplicate"): - to_chain(_scale, _scale) + to_chain(scale, scale) # empty rejected with pytest.raises(ValueError, match="at least one"): to_chain() From 7312d5aa1bc6de2e513c74f72d9e5b87e774ab31 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 16:59:39 +0100 Subject: [PATCH 14/20] Fix to_chain infra: use field default instead of post_init override Pass default infra through pydantic field validation so dicts are properly converted to Backend instances. Co-authored-by: Cursor --- exca/steps/base.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/exca/steps/base.py b/exca/steps/base.py index 8402fe0f..977e308d 100644 --- a/exca/steps/base.py +++ b/exca/steps/base.py @@ -598,7 +598,6 @@ def to_chain( step_classes = [to_step(f) for _, f in named] _field_names = tuple(name for name, _ in named) - _default_infra = infra # One field per step, typed as its Step subclass (with a default # instance so all fields are optional). @@ -610,6 +609,11 @@ def to_chain( # per-function fields in model_post_init. model_fields["steps"] = (tp.Sequence[Step], ()) + # Set the default infra via the field default so that pydantic + # validates it (converts dicts to Backend instances, etc.). + if infra is not None: + model_fields["infra"] = (backends.Backend | None, infra) + chain_name = "_".join(_field_names) + "_Chain" Model: tp.Type[Chain] = pydantic.create_model( @@ -625,8 +629,6 @@ def _model_post_init(self: tp.Any, __context: tp.Any) -> None: if not self.steps: built = tuple(getattr(self, name) for name in _field_names) object.__setattr__(self, "steps", built) - if _default_infra is not None and self.infra is None: - object.__setattr__(self, "infra", _default_infra) _super_post_init(self, __context) Model.model_post_init = _model_post_init # type: ignore[assignment] From cf6db3a45f47e6856638cd682d82e2b455c750f5 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 17:04:01 +0100 Subject: [PATCH 15/20] Fix to_chain infra: validate dict into Backend via model_validate The dict must go through Backend.model_validate to create a proper Backend instance before being assigned to the step. Co-authored-by: Cursor --- exca/steps/base.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/exca/steps/base.py b/exca/steps/base.py index 977e308d..2f99d6fc 100644 --- a/exca/steps/base.py +++ b/exca/steps/base.py @@ -609,10 +609,7 @@ def to_chain( # per-function fields in model_post_init. model_fields["steps"] = (tp.Sequence[Step], ()) - # Set the default infra via the field default so that pydantic - # validates it (converts dicts to Backend instances, etc.). - if infra is not None: - model_fields["infra"] = (backends.Backend | None, infra) + _default_infra = infra chain_name = "_".join(_field_names) + "_Chain" @@ -629,6 +626,10 @@ def _model_post_init(self: tp.Any, __context: tp.Any) -> None: if not self.steps: built = tuple(getattr(self, name) for name in _field_names) object.__setattr__(self, "steps", built) + if _default_infra is not None and self.infra is None: + infra_obj = backends.Backend.model_validate(_default_infra) + infra_obj._step = self + object.__setattr__(self, "infra", infra_obj) _super_post_init(self, __context) Model.model_post_init = _model_post_init # type: ignore[assignment] From 2711ca5518d26e65ed20b83934bb8957e1e9a76a Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 17:08:36 +0100 Subject: [PATCH 16/20] Fix black and isort formatting Co-authored-by: Cursor --- exca/steps/base.py | 18 +++++++++--------- exca/steps/test_steps.py | 12 +++++++++++- 2 files changed, 20 insertions(+), 10 deletions(-) diff --git a/exca/steps/base.py b/exca/steps/base.py index 2f99d6fc..215622dd 100644 --- a/exca/steps/base.py +++ b/exca/steps/base.py @@ -25,11 +25,12 @@ import exca from exca import utils -X = tp.TypeVar("X") - from . import backends from .backends import NoValue +X = tp.TypeVar("X") + + logger = logging.getLogger(__name__) @@ -441,9 +442,7 @@ def _resolve_input_params( sig = inspect.signature(func) if input_params is None: return tuple( - p.name - for p in sig.parameters.values() - if p.default is inspect._empty + p.name for p in sig.parameters.values() if p.default is inspect._empty ) for name in input_params: if name not in sig.parameters: @@ -495,21 +494,21 @@ def to_step( continue if p.name in reserved: raise ValueError( - f"Parameter {p.name!r} is reserved and cannot be used " - "as a field name" + f"Parameter {p.name!r} is reserved and cannot be used " "as a field name" ) if p.annotation in (tp.Any, inspect._empty): raise ValueError( - f"Parameter {p.name!r} of {func.__name__} needs a type " - "annotation" + f"Parameter {p.name!r} of {func.__name__} needs a type " "annotation" ) default = ... if p.default is inspect._empty else p.default fields[p.name] = (p.annotation, default) # Build _run that calls the original function if len(resolved) == 0: + def _run(self: tp.Any) -> tp.Any: # type: ignore[misc] return func(**{n: getattr(self, n) for n in fields}) + elif len(resolved) == 1: _input_name = resolved[0] @@ -517,6 +516,7 @@ def _run(self: tp.Any, value: tp.Any) -> tp.Any: # type: ignore[misc] kwargs = {n: getattr(self, n) for n in fields} kwargs[_input_name] = value return func(**kwargs) + else: _input_names = resolved diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index ceaef4aa..c821343a 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -7,6 +7,7 @@ """Tests for Step and Chain basic functionality (no caching tests here, see test_cache.py).""" import pickle +import random as _random import typing as tp from pathlib import Path @@ -318,7 +319,6 @@ def _forward(self, x: float) -> float: # to_step / to_chain helpers # ============================================================================= -import random as _random def generate(seed: int = 42) -> float: @@ -339,13 +339,17 @@ def test_to_step() -> None: M = to_step(scale) assert M(factor=3.0).run(5.0) == 15.0 # type: ignore[call-arg] assert M().run(5.0) == 50.0 + # explicit input_params override def add(a: float, b: float) -> float: return a + b + assert to_step(add, input_params=["a"])(b=10.0).run(5.0) == 15.0 # type: ignore[call-arg] + # multiple inputs -> tuple unpacking def combine(x: int, y: int, s: float = 1.0) -> float: return (x + y) * s + assert to_step(combine)(s=2.0).run((3, 7)) == 20.0 # type: ignore[call-arg] # in a Chain chain = Chain(steps=[G(seed=42), M(factor=100.0)]) # type: ignore[call-arg] @@ -363,14 +367,19 @@ def test_to_step_with_infra(tmp_path: Path) -> None: def test_to_step_validation_errors() -> None: def f(x: int) -> int: return x + with pytest.raises(ValueError, match="not a parameter"): to_step(f, input_params=["nope"]) + def g(x: int, y) -> int: # type: ignore[no-untyped-def] return x + y + with pytest.raises(ValueError, match="needs a type annotation"): to_step(g, input_params=["x"]) + def h(infra: int) -> int: return infra + with pytest.raises(ValueError, match="reserved"): to_step(h, input_params=[]) @@ -391,6 +400,7 @@ def test_to_chain() -> None: def test_to_chain_with_infra(tmp_path: Path) -> None: def double(x: float) -> float: return x * 2 + MyChain = to_chain(generate, double, infra={"backend": "Cached", "folder": tmp_path}) chain = MyChain() # type: ignore[call-arg] assert chain.run() == chain.run() From 2d94b8f6878d7677150c7e2ba561a52ffd311321 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 17:12:46 +0100 Subject: [PATCH 17/20] Remove extra blank line (black 24.3.0 compat) Co-authored-by: Cursor --- exca/steps/test_steps.py | 1 - 1 file changed, 1 deletion(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index c821343a..e0990fee 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -320,7 +320,6 @@ def _forward(self, x: float) -> float: # ============================================================================= - def generate(seed: int = 42) -> float: return _random.Random(seed).random() From db228c2717acf512a87ec8ae4ff9944730271075 Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 20:48:45 +0100 Subject: [PATCH 18/20] Simplify to_step/to_chain tests: merge into two test functions Remove test_to_step_with_infra, test_to_chain_with_infra, test_to_step_validation_errors, test_to_chain_named_and_errors. Fold their assertions into test_to_step and test_to_chain. Also rename import random as _random -> import random. Co-authored-by: Cursor --- exca/steps/test_steps.py | 59 ++++++++++++++-------------------------- 1 file changed, 21 insertions(+), 38 deletions(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index e0990fee..4b3a93ad 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -7,7 +7,7 @@ """Tests for Step and Chain basic functionality (no caching tests here, see test_cache.py).""" import pickle -import random as _random +import random import typing as tp from pathlib import Path @@ -321,7 +321,7 @@ def _forward(self, x: float) -> float: def generate(seed: int = 42) -> float: - return _random.Random(seed).random() + return random.Random(seed).random() def scale(x: float, factor: float = 10.0) -> float: @@ -338,13 +338,12 @@ def test_to_step() -> None: M = to_step(scale) assert M(factor=3.0).run(5.0) == 15.0 # type: ignore[call-arg] assert M().run(5.0) == 50.0 - # explicit input_params override def add(a: float, b: float) -> float: return a + b - assert to_step(add, input_params=["a"])(b=10.0).run(5.0) == 15.0 # type: ignore[call-arg] - + AddStep = to_step(add, input_params=["a"]) + assert AddStep(b=10.0).run(5.0) == 15.0 # type: ignore[call-arg] # multiple inputs -> tuple unpacking def combine(x: int, y: int, s: float = 1.0) -> float: return (x + y) * s @@ -352,18 +351,8 @@ def combine(x: int, y: int, s: float = 1.0) -> float: assert to_step(combine)(s=2.0).run((3, 7)) == 20.0 # type: ignore[call-arg] # in a Chain chain = Chain(steps=[G(seed=42), M(factor=100.0)]) # type: ignore[call-arg] - assert chain.run() == pytest.approx(_random.Random(42).random() * 100.0) - - -def test_to_step_with_infra(tmp_path: Path) -> None: - G = to_step(generate) - infra: tp.Any = {"backend": "Cached", "folder": tmp_path} - step = G(seed=7, infra=infra) # type: ignore[call-arg] - assert step.run() == step.run() - assert step.with_input().has_cache() - - -def test_to_step_validation_errors() -> None: + assert chain.run() == pytest.approx(random.Random(42).random() * 100.0) + # validation errors def f(x: int) -> int: return x @@ -387,29 +376,23 @@ def test_to_chain() -> None: MyChain = to_chain(generate, scale) assert issubclass(MyChain, Chain) # defaults - assert MyChain().run() == pytest.approx(_random.Random(42).random() * 10.0) # type: ignore[call-arg] - # custom - c = MyChain(generate=dict(seed=123), scale=dict(factor=100.0)) # type: ignore[call-arg] - assert c.run() == pytest.approx(_random.Random(123).random() * 100.0) - # partial + assert MyChain().run() == pytest.approx( # type: ignore[call-arg] + random.Random(42).random() * 10.0 + ) + # custom params via dict + c = MyChain( # type: ignore[call-arg] + generate=dict(seed=123), scale=dict(factor=100.0) + ) + assert c.run() == pytest.approx(random.Random(123).random() * 100.0) + # partial override c2 = MyChain(scale=dict(factor=5.0)) # type: ignore[call-arg] - assert c2.run() == pytest.approx(_random.Random(42).random() * 5.0) - - -def test_to_chain_with_infra(tmp_path: Path) -> None: - def double(x: float) -> float: - return x * 2 - - MyChain = to_chain(generate, double, infra={"backend": "Cached", "folder": tmp_path}) - chain = MyChain() # type: ignore[call-arg] - assert chain.run() == chain.run() - - -def test_to_chain_named_and_errors() -> None: + assert c2.run() == pytest.approx(random.Random(42).random() * 5.0) # (name, func) tuples for duplicate functions - MyChain = to_chain(generate, ("up", scale), ("down", scale)) - c = MyChain(up=dict(factor=100.0), down=dict(factor=0.5)) # type: ignore[call-arg] - assert c.run() == pytest.approx(generate() * 100.0 * 0.5) + MyChain2 = to_chain(generate, ("up", scale), ("down", scale)) + c3 = MyChain2( # type: ignore[call-arg] + up=dict(factor=100.0), down=dict(factor=0.5) + ) + assert c3.run() == pytest.approx(generate() * 100.0 * 0.5) # duplicate bare names rejected with pytest.raises(ValueError, match="Duplicate"): to_chain(scale, scale) From d2bd8e33dd569842f2e9caf5433a2366a90f94aa Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 20:52:45 +0100 Subject: [PATCH 19/20] Fix mypy: rename variable g -> gen to avoid redefinition Co-authored-by: Cursor --- exca/steps/test_steps.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index 4b3a93ad..5d4a244f 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -331,9 +331,9 @@ def scale(x: float, factor: float = 10.0) -> float: def test_to_step() -> None: # generator (all defaults) G = to_step(generate) - g = G(seed=123) # type: ignore[call-arg] - assert g._is_generator() - assert g.run() == g.run() + gen = G(seed=123) # type: ignore[call-arg] + assert gen._is_generator() + assert gen.run() == gen.run() # transformer (auto-detect required param as input) M = to_step(scale) assert M(factor=3.0).run(5.0) == 15.0 # type: ignore[call-arg] From cd9dd7e048a8b76a9e6e0f9de6de8cecf640f5cd Mon Sep 17 00:00:00 2001 From: kingjr Date: Wed, 25 Feb 2026 20:57:02 +0100 Subject: [PATCH 20/20] Fix black formatting (24.3.0) Co-authored-by: Cursor --- exca/steps/test_steps.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/exca/steps/test_steps.py b/exca/steps/test_steps.py index 5d4a244f..35aa953d 100644 --- a/exca/steps/test_steps.py +++ b/exca/steps/test_steps.py @@ -338,12 +338,14 @@ def test_to_step() -> None: M = to_step(scale) assert M(factor=3.0).run(5.0) == 15.0 # type: ignore[call-arg] assert M().run(5.0) == 50.0 + # explicit input_params override def add(a: float, b: float) -> float: return a + b AddStep = to_step(add, input_params=["a"]) assert AddStep(b=10.0).run(5.0) == 15.0 # type: ignore[call-arg] + # multiple inputs -> tuple unpacking def combine(x: int, y: int, s: float = 1.0) -> float: return (x + y) * s @@ -352,6 +354,7 @@ def combine(x: int, y: int, s: float = 1.0) -> float: # in a Chain chain = Chain(steps=[G(seed=42), M(factor=100.0)]) # type: ignore[call-arg] assert chain.run() == pytest.approx(random.Random(42).random() * 100.0) + # validation errors def f(x: int) -> int: return x @@ -389,9 +392,7 @@ def test_to_chain() -> None: assert c2.run() == pytest.approx(random.Random(42).random() * 5.0) # (name, func) tuples for duplicate functions MyChain2 = to_chain(generate, ("up", scale), ("down", scale)) - c3 = MyChain2( # type: ignore[call-arg] - up=dict(factor=100.0), down=dict(factor=0.5) - ) + c3 = MyChain2(up=dict(factor=100.0), down=dict(factor=0.5)) # type: ignore[call-arg] assert c3.run() == pytest.approx(generate() * 100.0 * 0.5) # duplicate bare names rejected with pytest.raises(ValueError, match="Duplicate"):