psc_NN (Machine Learning for Perovskite Solar Cells: An Open-Source Pipeline)

Notes

1. Before starting training, please ensure that the dataset has been correctly placed in the data/cleaned/ directory.
2. Training and evaluation require additional dependencies, please install them using pip install -r requirements.txt.
3. For optimal performance, it is recommended to use GPU for training.

Model Training CommandModel Evaluation Command

python psc_nn.py

# Use local pre-trained model
wget -c https://paddle-org.bj.bcebos.com/paddlescience/datasets/psc/data.zip
unzip data.zip
python psc_nn.py mode=eval eval.pretrained_model_path="Your pdparams path"

# Or use provided pre-trained model
wget -c https://paddle-org.bj.bcebos.com/paddlescience/datasets/psc/data.zip
unzip data.zip
python psc_nn.py mode=eval eval.pretrained_model_path="https://paddle-org.bj.bcebos.com/paddlescience/models/PerovskiteSolarCells/solar_cell_pretrained.pdparams"

Pretrained Model	Metrics
solar_cell_pretrained.pdparams	RMSE: 3.91798

1. Background Introduction

Solar cells are key energy devices that directly convert light energy into electrical energy through the photovoltaic effect. Performance prediction is an important part of optimizing and designing solar cells. However, traditional performance prediction methods often rely on complex physical simulations and a large number of experimental tests, which are not only costly but also time-consuming, restricting the efficiency of research and development.

In recent years, the rapid development of deep learning and machine learning technologies has provided innovative methods for solar cell performance prediction. Through machine learning technology, development speed can be significantly accelerated while achieving prediction accuracy comparable to experimental results. Especially in the research of perovskite solar cells, the chemical composition and structural diversity of materials bring new challenges to model training. To solve this problem, researchers usually convert material properties into fixed-length feature vectors to adapt to machine learning models. Nevertheless, the feature representation design for different performance indicators still needs continuous optimization, and the interpretability requirements for model prediction results are also stricter.

In this study, by utilizing a comprehensive database (PDP) containing information on the properties of perovskite solar cells, we constructed and evaluated a variety of machine learning models including XGBoost and psc_nn, focusing on predicting short-circuit current density (Jsc). The results show that combining deep learning with hyperparameter optimization tools (such as Optuna) can significantly improve the efficiency of solar cell design, providing a more accurate and efficient solution for the research and development of new solar cells.

2. Model Principle

This chapter only briefly introduces the principle of the solar cell performance prediction model. For detailed theoretical derivation, please read Machine Learning for Perovskite Solar Cells: An Open-Source Pipeline.

The main idea of this method is to establish a nonlinear mapping relationship between spectral response data and short-circuit current density (Jsc) through an artificial neural network. The overall structure of the artificial neural network model is shown in the figure below:

This case uses a Multi-Layer Perceptron (MLP) as the basic model architecture, mainly including the following parts:

1. Input layer: Receives 2808-dimensional spectral response data
2. Hidden layer: 4-6 fully connected layers, the number of neurons in each layer is optimized by Optuna
3. Activation function: Uses ReLU activation function to introduce nonlinear characteristics
4. Output layer: Outputs the predicted Jsc value

In this way, we can automatically find the model configuration best suited for the current task and improve the prediction performance of the model.

3. Model Implementation

In this chapter, we explain how to implement the perovskite solar cell performance prediction model based on PaddleScience code. This case combines the Optuna framework for hyperparameter optimization and uses various built-in functional modules of PaddleScience. In order to quickly understand PaddleScience, only key steps such as model construction, constraint construction, and validator construction are described below, while other details please refer to API Documentation.

3.1 Dataset Introduction

The dataset used in this case contains Perovskite Database Project (PDP) data. The dataset is divided into the following parts:

1. Training set:
2. Feature data: data/cleaned/training.csv
3. Label data: data/cleaned/training_labels.csv
4. Validation set:
5. Feature data: data/cleaned/validation.csv
6. Label data: data/cleaned/validation_labels.csv

To facilitate data processing, we implemented a helper function create_tensor_dict to create a tensor dictionary of inputs and labels:

def create_tensor_dict(X, y):
    """Create Tensor Dictionary for Input and Labels"""
    return {
        "input": paddle.to_tensor(X.values, dtype="float32"),
        "label": {"target": paddle.to_tensor(y.values, dtype="float32")},
    }

The data reading and preprocessing code is as follows:

# Read and preprocess data
X_train = pd.read_csv(cfg.data.train_features_path)
y_train = pd.read_csv(cfg.data.train_labels_path)
X_val = pd.read_csv(cfg.data.val_features_path)
y_val = pd.read_csv(cfg.data.val_labels_path)

for col in X_train.columns:
    if "[" in col or "]" in col:
        old_name = col
        new_name = col.replace("[", "(").replace("]", ")")
        X_train = X_train.rename(columns={old_name: new_name})
        X_val = X_val.rename(columns={old_name: new_name})

X_train, X_verif, y_train, y_verif = train_test_split(
    X_train, y_train, test_size=0.1, random_state=42
)

for df in [X_train, y_train, X_verif, y_verif, X_val, y_val]:
    df.reset_index(drop=True, inplace=True)

For hyperparameter optimization, we further divide the training set into training set and validation set:

X_train, X_verif, y_train, y_verif = train_test_split(
    X_train, y_train, test_size=0.1, random_state=42
)

3.2 Model Construction

This case uses ppsci.arch.MLP built in PaddleScience to build a multi-layer perceptron model. The hyperparameters of the model are optimized through the Optuna framework, mainly including:

1. Number of network layers: 4-6 layers
2. Number of neurons per layer: 10-input_dim/2
3. Activation function: ReLU
4. Input dimension: 2808 (spectral response data dimension)
5. Output dimension: 1 (Jsc predicted value)

The model definition code is as follows:

def define_model(trial, input_dim, output_dim):
    n_layers = trial.suggest_int("n_layers", 4, 6)
    hidden_sizes = []
    for i in range(n_layers):
        out_features = trial.suggest_int(f"n_units_l{i}", 10, input_dim // 2)
        hidden_sizes.append(out_features)

    model = ppsci.arch.MLP(
        input_keys=("input",),
        output_keys=("target",),
        num_layers=None,
        hidden_size=hidden_sizes,
        activation="relu",
        input_dim=input_dim,
        output_dim=output_dim,
    )
    return model

3.3 Loss Function Design

Considering that different samples in the dataset may have different importance, we designed a weighted mean square error loss function. This function assigns higher weight to larger Jsc values to improve the prediction accuracy of the model on high-performance solar cells:

def weighted_loss(output_dict, target_dict, weight_dict=None):
    pred = output_dict["target"]
    true = target_dict["target"]
    epsilon = 1e-06
    n = len(true)
    weights = true / (paddle.sum(x=true) + epsilon)
    squared = (true - pred) ** 2
    weighted = squared * weights
    loss = paddle.sum(x=weighted) / n
    return {"weighted_mse": loss}

3.4 Constraint Construction

This case solves the problem based on data-driven methods, so SupervisedConstraint built in PaddleScience is used to construct supervised constraints. To reduce code duplication, we implemented the create_constraint function to create supervised constraints:

def create_constraint(input_dict, batch_size, shuffle=True):
    """Create supervision constraints"""
    return SupervisedConstraint(
        dataloader_cfg={
            "dataset": {
                "name": "NamedArrayDataset",
                "input": {"input": input_dict["input"]},
                "label": input_dict["label"],
            },
            "batch_size": batch_size,
            "sampler": {
                "name": "BatchSampler",
                "drop_last": False,
                "shuffle": shuffle,
            },
        },
        loss=weighted_loss,
        output_expr={"target": lambda out: out["target"]},
        name="train_constraint",
    )

3.5 Validator Construction

In order to monitor the training situation of the model in real time, we implemented the create_validator function to create a validator:

def create_validator(input_dict, batch_size, name="validator"):
    """Create an evaluator"""
    return SupervisedValidator(
        dataloader_cfg={
            "dataset": {
                "name": "NamedArrayDataset",
                "input": {"input": input_dict["input"]},
                "label": input_dict["label"],
            },
            "batch_size": batch_size,
        },
        loss=weighted_loss,
        output_expr={"target": lambda out: out["target"]},
        metric={"RMSE": ppsci.metric.RMSE(), "MAE": ppsci.metric.MAE()},
        name=name,
    )

3.6 Optimizer Construction

In order to unify the management of the creation of optimizer and learning rate scheduler, we implemented the create_optimizer function:

def create_optimizer(model, optimizer_name, lr, train_cfg, data_size):
    """Create optimizer and learning rate scheduler"""
    schedule = lr_scheduler.ExponentialDecay(
        epochs=train_cfg.epochs,
        iters_per_epoch=data_size // train_cfg.batch_size,
        learning_rate=lr,
        gamma=train_cfg.lr_scheduler.gamma,
        decay_steps=train_cfg.lr_scheduler.decay_steps,
        warmup_epoch=train_cfg.lr_scheduler.warmup_epoch,
        warmup_start_lr=train_cfg.lr_scheduler.warmup_start_lr,
    )()

    if optimizer_name == "Adam":
        return optimizer.Adam(learning_rate=schedule)(model)
    elif optimizer_name == "RMSProp":
        return optimizer.RMSProp(learning_rate=schedule)(model)
    else:
        return optimizer.SGD(learning_rate=schedule)(model)

3.7 Model Training and Evaluation

During the training process, we use the functions encapsulated above to create data dictionaries, constraints, validators and optimizers:

train_dict = create_tensor_dict(X_train, y_train)
val_dict = create_tensor_dict(X_val, y_val)

train_constraint = create_constraint(train_dict, cfg.TRAIN.batch_size)
val_validator = create_validator(val_dict, cfg.EVAL.batch_size, "val_validator")

4. Complete Code

import os
import tempfile
import urllib.request
from os import path as osp

import hydra
import numpy as np
import optuna
import paddle
import pandas as pd
from matplotlib import pyplot as plt
from omegaconf import DictConfig
from sklearn.metrics import mean_absolute_percentage_error
from sklearn.metrics import mean_squared_error
from sklearn.metrics import r2_score
from sklearn.model_selection import train_test_split

import ppsci
from ppsci.constraint import SupervisedConstraint
from ppsci.optimizer import lr_scheduler
from ppsci.optimizer import optimizer
from ppsci.solver import Solver
from ppsci.validate import SupervisedValidator


def download_model_from_url(url, local_path=None):
    """
    Download model from URL to local path.

    Args:
        url (str): URL to download the model from
        local_path (str, optional): Local path to save the model.
                                   If None, saves to a temporary file.

    Returns:
        str: Path to the downloaded model file
    """
    if local_path is None:
        # Create a temporary file
        temp_dir = tempfile.gettempdir()
        local_path = os.path.join(temp_dir, "downloaded_model.pdparams")

    print(f"Downloading model from {url} to {local_path}")

    try:
        urllib.request.urlretrieve(url, local_path)
        print(f"Successfully downloaded model to {local_path}")
        return local_path
    except Exception as e:
        raise RuntimeError(f"Failed to download model from {url}: {str(e)}")


def load_model_from_path_or_url(model_path):
    """
    Load model from local path or URL.

    Args:
        model_path (str): Local path or URL to the model

    Returns:
        dict: Loaded model dictionary
    """
    if model_path.startswith(("http://", "https://")):
        # It's a URL, download first
        local_path = download_model_from_url(model_path)
        return paddle.load(local_path)
    else:
        # It's a local path
        return paddle.load(model_path)


def weighted_loss(output_dict, target_dict, weight_dict=None):
    pred = output_dict["target"]
    true = target_dict["target"]
    epsilon = 1e-06
    n = len(true)
    weights = true / (paddle.sum(x=true) + epsilon)
    squared = (true - pred) ** 2
    weighted = squared * weights
    loss = paddle.sum(x=weighted) / n
    return {"weighted_mse": loss}


def create_tensor_dict(X, y):
    """Create Tensor Dictionary for Input and Labels"""
    return {
        "input": paddle.to_tensor(X.values, dtype="float32"),
        "label": {"target": paddle.to_tensor(y.values, dtype="float32")},
    }


def create_constraint(input_dict, batch_size, shuffle=True):
    """Create supervision constraints"""
    return SupervisedConstraint(
        dataloader_cfg={
            "dataset": {
                "name": "NamedArrayDataset",
                "input": {"input": input_dict["input"]},
                "label": input_dict["label"],
            },
            "batch_size": batch_size,
            "sampler": {
                "name": "BatchSampler",
                "drop_last": False,
                "shuffle": shuffle,
            },
        },
        loss=weighted_loss,
        output_expr={"target": lambda out: out["target"]},
        name="train_constraint",
    )


def create_validator(input_dict, batch_size, name="validator"):
    """Create an evaluator"""
    return SupervisedValidator(
        dataloader_cfg={
            "dataset": {
                "name": "NamedArrayDataset",
                "input": {"input": input_dict["input"]},
                "label": input_dict["label"],
            },
            "batch_size": batch_size,
        },
        loss=weighted_loss,
        output_expr={"target": lambda out: out["target"]},
        metric={"RMSE": ppsci.metric.RMSE(), "MAE": ppsci.metric.MAE()},
        name=name,
    )


def create_optimizer(model, optimizer_name, lr, train_cfg, data_size):
    """Create optimizer and learning rate scheduler"""
    schedule = lr_scheduler.ExponentialDecay(
        epochs=train_cfg.epochs,
        iters_per_epoch=data_size // train_cfg.batch_size,
        learning_rate=lr,
        gamma=train_cfg.lr_scheduler.gamma,
        decay_steps=train_cfg.lr_scheduler.decay_steps,
        warmup_epoch=train_cfg.lr_scheduler.warmup_epoch,
        warmup_start_lr=train_cfg.lr_scheduler.warmup_start_lr,
    )()

    if optimizer_name == "Adam":
        return optimizer.Adam(learning_rate=schedule)(model)
    elif optimizer_name == "RMSProp":
        return optimizer.RMSProp(learning_rate=schedule)(model)
    else:
        return optimizer.SGD(learning_rate=schedule)(model)


def define_model(trial, input_dim, output_dim):
    n_layers = trial.suggest_int("n_layers", 4, 6)
    hidden_sizes = []
    for i in range(n_layers):
        out_features = trial.suggest_int(f"n_units_l{i}", 10, input_dim // 2)
        hidden_sizes.append(out_features)

    model = ppsci.arch.MLP(
        input_keys=("input",),
        output_keys=("target",),
        num_layers=None,
        hidden_size=hidden_sizes,
        activation="relu",
        input_dim=input_dim,
        output_dim=output_dim,
    )
    return model


def train(cfg: DictConfig):
    # Read and preprocess data
    X_train = pd.read_csv(cfg.data.train_features_path)
    y_train = pd.read_csv(cfg.data.train_labels_path)
    X_val = pd.read_csv(cfg.data.val_features_path)
    y_val = pd.read_csv(cfg.data.val_labels_path)

    for col in X_train.columns:
        if "[" in col or "]" in col:
            old_name = col
            new_name = col.replace("[", "(").replace("]", ")")
            X_train = X_train.rename(columns={old_name: new_name})
            X_val = X_val.rename(columns={old_name: new_name})

    X_train, X_verif, y_train, y_verif = train_test_split(
        X_train, y_train, test_size=0.1, random_state=42
    )

    for df in [X_train, y_train, X_verif, y_verif, X_val, y_val]:
        df.reset_index(drop=True, inplace=True)

    def objective(trial):
        model = define_model(trial, cfg.model.input_dim, cfg.model.output_dim)

        optimizer_name = trial.suggest_categorical(
            "optimizer", ["Adam", "RMSProp", "SGD"]
        )
        lr = trial.suggest_float("lr", 1e-5, 1e-1, log=True)

        train_dict = create_tensor_dict(X_train, y_train)
        verif_dict = create_tensor_dict(X_verif, y_verif)

        opt = create_optimizer(model, optimizer_name, lr, cfg.TRAIN, len(X_train))

        train_constraint = create_constraint(train_dict, cfg.TRAIN.batch_size)
        verif_validator = create_validator(
            verif_dict, cfg.EVAL.batch_size, "verif_validator"
        )

        solver = Solver(
            model=model,
            constraint={"train": train_constraint},
            optimizer=opt,
            validator={"verif": verif_validator},
            output_dir=cfg.output_dir,
            epochs=cfg.TRAIN.search_epochs,
            iters_per_epoch=len(X_train) // cfg.TRAIN.batch_size,
            eval_during_train=cfg.TRAIN.eval_during_train,
            eval_freq=cfg.TRAIN.eval_freq,
            save_freq=cfg.TRAIN.save_freq,
            eval_with_no_grad=cfg.EVAL.eval_with_no_grad,
            log_freq=cfg.TRAIN.log_freq,
        )

        solver.train()

        verif_preds = solver.predict({"input": verif_dict["input"]}, return_numpy=True)[
            "target"
        ]

        verif_rmse = np.sqrt(mean_squared_error(y_verif.values, verif_preds))

        return verif_rmse

    study = optuna.create_study()
    study.optimize(objective, n_trials=50)

    best_params = study.best_trial.params
    print("\nBest hyperparameters: " + str(best_params))

    # Save the optimal model structure
    hidden_sizes = []
    for i in range(best_params["n_layers"]):
        hidden_sizes.append(best_params[f"n_units_l{i}"])

    # Create and train the final model
    final_model = define_model(
        study.best_trial, cfg.model.input_dim, cfg.model.output_dim
    )
    opt = create_optimizer(
        final_model,
        best_params["optimizer"],
        best_params["lr"],
        cfg.TRAIN,
        len(X_train),
    )

    train_dict = create_tensor_dict(X_train, y_train)
    val_dict = create_tensor_dict(X_val, y_val)

    train_constraint = create_constraint(train_dict, cfg.TRAIN.batch_size)
    val_validator = create_validator(val_dict, cfg.EVAL.batch_size, "val_validator")

    solver = Solver(
        model=final_model,
        constraint={"train": train_constraint},
        optimizer=opt,
        validator={"valid": val_validator},
        output_dir=cfg.output_dir,
        epochs=cfg.TRAIN.epochs,
        iters_per_epoch=len(X_train) // cfg.TRAIN.batch_size,
        eval_during_train=cfg.TRAIN.eval_during_train,
        eval_freq=cfg.TRAIN.eval_freq,
        save_freq=cfg.TRAIN.save_freq,
        eval_with_no_grad=cfg.EVAL.eval_with_no_grad,
        log_freq=cfg.TRAIN.log_freq,
    )

    solver.train()

    # Save model structure and weights
    model_dict = {
        "state_dict": final_model.state_dict(),
        "hidden_size": hidden_sizes,
        "n_layers": best_params["n_layers"],
        "optimizer": best_params["optimizer"],
        "lr": best_params["lr"],
    }
    paddle.save(
        model_dict, os.path.join(cfg.output_dir, "checkpoints", "best_model.pdparams")
    )
    print(
        "Saved model structure and weights to "
        + os.path.join(cfg.output_dir, "checkpoints", "best_model.pdparams")
    )

    solver.plot_loss_history(by_epoch=True, smooth_step=1)
    solver.eval()

    visualize_results(solver, X_val, y_val, cfg.output_dir)


def evaluate(cfg: DictConfig):
    # Read and preprocess data
    X_val = pd.read_csv(cfg.data.val_features_path)
    y_val = pd.read_csv(cfg.data.val_labels_path)

    for col in X_val.columns:
        if "[" in col or "]" in col:
            old_name = col
            new_name = col.replace("[", "(").replace("]", ")")
            X_val = X_val.rename(columns={old_name: new_name})

    # Loading model structure and weights
    print(f"Loading model from {cfg.EVAL.pretrained_model_path}")
    model_dict = load_model_from_path_or_url(cfg.EVAL.pretrained_model_path)
    hidden_size = model_dict["hidden_size"]
    print(f"Loaded model structure with hidden sizes: {hidden_size}")

    model = ppsci.arch.MLP(
        input_keys=("input",),
        output_keys=("target",),
        num_layers=None,
        hidden_size=hidden_size,
        activation="relu",
        input_dim=cfg.model.input_dim,
        output_dim=cfg.model.output_dim,
    )

    # Load model weights
    model.set_state_dict(model_dict["state_dict"])
    print("Successfully loaded model weights")

    valid_dict = create_tensor_dict(X_val, y_val)
    valid_validator = create_validator(
        valid_dict, cfg.EVAL.batch_size, "valid_validator"
    )

    solver = Solver(
        model=model,
        output_dir=cfg.output_dir,
        validator={"valid": valid_validator},
        eval_with_no_grad=cfg.EVAL.eval_with_no_grad,
    )

    # evaluation model
    print("Evaluating model...")
    solver.eval()

    # Generate prediction results
    predictions = solver.predict({"input": valid_dict["input"]}, return_numpy=True)[
        "target"
    ]

    # Calculate multiple evaluation indicators
    rmse = np.sqrt(mean_squared_error(y_val.values, predictions))
    r2 = r2_score(y_val.values, predictions)
    mape = mean_absolute_percentage_error(y_val.values, predictions)

    print("Evaluation metrics:")
    print(f"RMSE: {rmse:.5f}")
    print(f"R2 Score: {r2:.5f}")
    print(f"MAPE: {mape:.5f}")

    # Visualization results
    print("Generating visualization...")
    visualize_results(solver, X_val, y_val, cfg.output_dir)
    print("Evaluation completed.")


def visualize_results(solver, X_val, y_val, output_dir):
    pred_dict = solver.predict(
        {"input": paddle.to_tensor(X_val.values, dtype="float32")}, return_numpy=True
    )
    val_preds = pred_dict["target"]
    val_true = y_val.values

    plt.figure(figsize=(10, 6))
    plt.grid(True, linestyle="--", alpha=0.7)
    plt.hist(val_true, bins=30, alpha=0.6, label="True Jsc", color="tab:blue")
    plt.hist(val_preds, bins=30, alpha=0.6, label="Predicted Jsc", color="orange")

    pred_mean = np.mean(val_preds)
    pred_std = np.std(val_preds)
    plt.axvline(pred_mean, color="black", linestyle="--")
    plt.axvline(pred_mean + pred_std, color="red", linestyle="--")
    plt.axvline(pred_mean - pred_std, color="red", linestyle="--")

    val_rmse = np.sqrt(mean_squared_error(val_true, val_preds))
    plt.title(f"Distribution of True Jsc vs Pred Jsc: RMSE {val_rmse:.5f}", pad=20)
    plt.xlabel("Jsc (mA/cm²)")
    plt.ylabel("Counts")
    plt.legend(fontsize=10)
    plt.tight_layout()
    plt.savefig(
        osp.join(output_dir, "jsc_distribution.png"), dpi=300, bbox_inches="tight"
    )
    plt.close()


@hydra.main(version_base=None, config_path="./conf", config_name="psc_nn.yaml")
def main(cfg: DictConfig):
    if cfg.mode == "train":
        train(cfg)
    elif cfg.mode == "eval":
        evaluate(cfg)
    else:
        raise ValueError(f"cfg.mode should in ['train', 'eval'], but got '{cfg.mode}'")


if __name__ == "__main__":
    main()

5. References

• Machine Learning for Perovskite Solar Cells: An Open-Source Pipeline