KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use parameters_source method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

loader.py

Source:loader.py

1import yaml2import json3import re4from abc import ABCMeta, abstractmethod5import six6import os7from cerberus import Validator8from collections import namedtuple9class InvalidDataException(Exception):10    def __init__(self, errors):11        self.errors = errors12def dict_to_namedtuple(dictionary):13    for key, value in dictionary.items():14            if isinstance(value, dict):15                dictionary[key] = dict_to_namedtuple(value)16            elif isinstance(value, list):17                el = []18                for i in value:19                    if isinstance(i, dict):20                        el.append(dict_to_namedtuple(i))21                    else:22                        el.append(i)23                dictionary[key] = el24    return namedtuple('Configuration', dictionary.keys())(**dictionary)25def is_string(value):26    try:27        float(value)28        return False29    except ValueError:30        if value.lower() in ["true", "false"]:31            return False32        else:33            return True34type_map = {35    int: "integer",36    str: "string",37    float: "float",38    bool: "boolean"39}40def mapped_to_cerberus(d):41    for key, value in d.items():42        if value is None:43            d[key] = {"required": True}44        elif isinstance(value, dict):45            if "type" not in value:46                schema_dict = {"type": "dict", "default": {}, "schema": mapped_to_cerberus(value)}47                d[key] = schema_dict48        elif isinstance(value, list):49            if len(value) > 0:50                schema_list = mapped_to_cerberus({"_":value[0]})["_"]51            else:52                schema_list = {}53            if isinstance(value, list):54                if isinstance(value[0], dict):55                    default_value = []56                else:57                    default_value = value58            elif not isinstance(value, dict):59                default_value = value60            else:61                default_value = []62            d[key] = {"type": "list", "default": default_value, "schema": schema_list}63        else:64                new_value = {65                    "type": type_map[type(value)],66                    "default": value,67                    "required": False68                }69                d[key] = new_value70    return d71def build_config(mapped_schema, config_data, as_named_tuple=True):72    if not isinstance(mapped_schema, dict):73        mapped_schema = mapped_schema.build()74    cer = mapped_to_cerberus(mapped_schema)75    v = Validator(cer)76    if v.validate(config_data):77        if as_named_tuple:78            config = dict_to_namedtuple(v.document)79        else:80            config = v.document81        return config82    else:83        raise InvalidDataException(format_errors(v.errors))84def format_errors(errors, prefix=[]):85    error_list = []86    for key, value in errors.items():87        for err in value:88            node_prefix = prefix + [str(key)]89            if isinstance(err, dict):90                error_list += format_errors(err, node_prefix)91            else:92                error_list.append("Field [{field}] {desc}".format(field=":".join(node_prefix), desc=err))93    return error_list94@six.add_metaclass(ABCMeta)95class ConfigurationLoader(object):96    @abstractmethod97    def load_parameters(self, source):98        """Convert the source into a dictionary"""99        pass100    @abstractmethod101    def load_config(self, config_source, parameters_source):102        pass103    def build_config(self, data, mapping, as_namedtuple=True):104        return build_config(mapped_schema=mapping, config_data=data, as_named_tuple=as_namedtuple)105class YmlLoader(ConfigurationLoader):106    def load_parameters(self, source):107        """For YML, the source it the file path"""108        with open(source) as parameters_source:109            loaded = yaml.safe_load(parameters_source.read())110            for k, v in loaded.items():111                if isinstance(v, str):112                    loaded[k] = "'"+v+"'"113            return loaded114    def load_config(self, config_source, parameters_source):115        """For YML, the source it the file path"""116        with open(config_source) as config_source:117            config_raw = config_source.read()118            parameters = {}119            """Parameteres from file"""120            if os.path.isfile(parameters_source):121                params = self.load_parameters(parameters_source)122                if params is not None:123                    parameters.update(params)124            """Overwrite parameteres with the environment variables"""125            env_params = {}126            env_params.update(os.environ)127            for k, v in env_params.items():128                if is_string(v):129                    env_params[k] = "'" + v + "'"130            parameters.update(env_params)131            """Replace the parameters"""132            final_configuration = config_raw.format(**parameters)133            final_configuration = yaml.safe_load(final_configuration)134            return final_configuration if final_configuration is not None else {}135class JsonLoader(ConfigurationLoader):136    def __init__(self):137        self.parameters = None138    def load_parameters(self, source):139        """For JSON, the source it the file path"""140        with open(source) as parameters_source:141            return json.loads(parameters_source.read())142    def load_config(self, config_source, parameters_source):143        """For JSON, the source it the file path"""144        with open(config_source) as config_source:145            config_raw = config_source.read()146            """Replace the parameters"""147            pattern = "(%[a-zA-Z_0-9]*%)"148            self.parameters = {}149            """Parameteres from file"""150            if os.path.isfile(parameters_source):151                self.parameters.update(self.load_parameters(parameters_source))152            """Overwrite parameteres with the environment variables"""153            self.parameters.update(os.environ)154            replaced_config = re.sub(pattern=pattern, repl=self._replace_function, string=config_raw)155            return json.loads(replaced_config)156    def _replace_function(self, match):157        # Remove % from the begining and from the end158        parameter_key = match.group(0)[1:-1]159        value = self.parameters[parameter_key]160        # Add the " for string values...

source_check.py

Source:source_check.py

1import pytorch_lightning as pl2from torchmetrics import Accuracy, ConfusionMatrix, MeanMetric3import torch4import torch.optim.lr_scheduler as lr_sched5from torch.nn.functional import softmax, one_hot, cross_entropy6from typing import List, Optional7from src.model_utils import *8import logging 9import wandb10from src.core_utils import BBE_estimate_binary, get_label_dist, BBE_estimate_multiclass11log = logging.getLogger("app")12class SourceDiscriminator(pl.LightningModule):13    def __init__(14        self,15        arch: str = "Resnet18",16        num_source_classes: int = 10,17        dataset: str=  "CIFAR10",18        learning_rate: float = 0.1,19        weight_decay: float = 5e-4,20        max_epochs: int = 500,21        pred_save_path: str = ".",22        work_dir: str = ".",23        hash: Optional[str] = None,24        pretrained: bool = False,25    ):26        super().__init__()27        self.num_classes = num_source_classes28        self.criterion = torch.nn.CrossEntropyLoss()29        self.num_outputs = self.num_classes30        self.source_model = get_model(arch, dataset, self.num_outputs, pretrained= pretrained)31        self.learning_rate = learning_rate32        self.weight_decay = weight_decay33        self.max_epochs = max_epochs34        self.pred_save_path = pred_save_path35        self.work_dir = work_dir36        self.hash = hash37        self.pretrained = pretrained38        self.is_clip = arch.startswith("Clip")39        self.automatic_optimization = False40    def forward_source(self, x):41        return self.source_model(x)42    def process_batch(self, batch, stage="train"):43        44        if stage == "train": 45            x_s, y_s, = batch #["source_full"]46            # _ = batch["target_full"]47            48            source_opt = self.optimizers()49            sch1 = self.lr_schedulers()50            logits_source = self.forward_source(x_s)51            loss1 = cross_entropy(logits_source, y_s)52            # log.debug(f"Batch logits size {logits.shape} ")53            source_opt.zero_grad()54            self.manual_backward(loss1)55            source_opt.step()56            if self.trainer.is_last_batch:57                sch1.step()58            return loss159        elif stage == "pred_source":60            x_s, y_s, = batch61            logits_s = self.source_model(x_s)62            probs_s = softmax(logits_s, dim=1)63            return probs_s, y_s64        else: 65            raise ValueError("Invalid stage %s" % stage)66    def training_step(self, batch, batch_idx: int):67        loss1 = self.process_batch(batch, "train")68        self.log("train/loss", {"source" : loss1}, on_step=True, on_epoch=True, prog_bar=True)69        70        return  {"source_loss": loss1.detach()}71    def validation_step(self, batch, batch_idx: int, dataloader_idx: int = 0):72        73        if dataloader_idx == 0: 74            probs_s, y_s = self.process_batch(batch, "pred_source")75            return {"probs_s": probs_s, "y_s": y_s }76    def validation_epoch_end(self, outputs):77        probs_s = torch.cat([x["probs_s"] for x in outputs], dim=0).detach().cpu().numpy()78        y_s = torch.cat([x["y_s"] for x in outputs], dim=0).detach().cpu().numpy()79        pred_s = np.argmax(probs_s, axis=1)80        ood_idx = np.where(y_s == self.num_classes)[0]81        log.info(f"OOD samples {len(ood_idx)}")82        log.info(f"Accuracy {np.mean(pred_s == y_s)}")83    def configure_optimizers(self):84        if self.is_clip:85            parameters_source = self.source_model.linear.parameters()86        else:87            parameters_source = self.source_model.parameters()88        optimizer_source = torch.optim.SGD(89            parameters_source,90            lr=self.learning_rate,91            weight_decay=self.weight_decay,92            momentum=0.993        )...

configuration.py

Source:configuration.py

1"""Configuration format loaders"""2import locale3import os4from abc import ABC, abstractmethod5import yaml6from pydantic import create_model7def load_configuration(configuration_file_path, parameters_file_path, bundles):8    """Combines the configuration and parameters and build the configuration object"""9    mappings = {}10    for bundle in bundles:11        if hasattr(bundle, "config_mapping"):12            mappings.update(bundle.config_mapping)13    loader = YmlLoader()14    return loader.build_config(mappings, config_source=configuration_file_path, parameters_source=parameters_file_path)15def is_string(value):16    """Check if the value is actually a string or not"""17    try:18        float(value)19        return False20    except ValueError:21        if value.lower() in ["true", "false"]:22            return False23        return True24class ConfigurationLoader(ABC):25    """Base configuration loader"""26    @abstractmethod27    def load_parameters(self, source):28        """Convert the source into a dictionary"""29    @abstractmethod30    def load_config(self, config_source, parameters_source):31        """Prase the config file and build a dictionary"""32    def build_config(self, config_mappings, config_source, parameters_source):33        """By using the loaded parameters and loaded config, build the final configuration object"""34        configuration_class = create_model('Configuration', **{k: (v, ...) for k, v in config_mappings.items()})35        return configuration_class(**self.load_config(config_source, parameters_source))36class YmlLoader(ConfigurationLoader):37    """YML Format parser and config loader"""38    def load_parameters(self, source):39        """For YML, the source it the file path"""40        with open(source, encoding=locale.getpreferredencoding(False)) as parameters_source:41            loaded = yaml.safe_load(parameters_source.read())42            if loaded:43                for key, value in loaded.items():44                    if isinstance(value, str):45                        loaded[key] = "'" + value + "'"46                return loaded47        return {}48    def load_config(self, config_source, parameters_source):49        """For YML, the source it the file path"""50        with open(config_source, encoding=locale.getpreferredencoding(False)) as config_source_file:51            config_raw = config_source_file.read()52            parameters = {}53            # Parameters from file54            if os.path.isfile(parameters_source):55                params = self.load_parameters(parameters_source)56                if params is not None:57                    parameters.update(params)58            # Overwrite parameters with the environment variables59            env_params = {}60            env_params.update(os.environ)61            for key, value in env_params.items():62                if is_string(value):63                    env_params[key] = "'" + value + "'"64            parameters.update(env_params)65            # Replace the parameters66            final_configuration = config_raw.format(**parameters)67            final_configuration = yaml.safe_load(final_configuration)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.