Source code for astroNN.models

import importlib
import json
import os
import sys
import warnings
from packaging import version

import h5py
from astroNN.config import custom_model_path_reader
from astroNN.models.apogee_models import (
    ApogeeBCNN,
    ApogeeCVAE,
    ApogeeCNN,
    ApogeeBCNNCensored,
    ApogeeDR14GaiaDR2BCNN,
    ApogeeKplerEchelle,
    StarNet2017,
)
from astroNN.models.misc_models import Cifar10CNN, MNIST_BCNN, SimplePolyNN
from astroNN.nn.losses import losses_lookup
from astroNN.nn.utilities import Normalizer
from astroNN.shared.dict_tools import dict_list_to_dict_np, list_to_dict
from tensorflow import keras
import tensorflow as tf

__all__ = [
    "load_folder",
    "ApogeeBCNN",
    "ApogeeCVAE",
    "ApogeeCNN",
    "ApogeeBCNNCensored",
    "ApogeeDR14GaiaDR2BCNN",
    "ApogeeKplerEchelle",
    "StarNet2017",
    "Cifar10CNN",
    "MNIST_BCNN",
    "SimplePolyNN",
]

optimizers = keras.optimizers
Sequential = keras.models.Sequential


def Galaxy10CNN():
    """
    NAME:
        Galaxy10CNN
    PURPOSE:
        setup Galaxy10CNN from Cifar10CNN with Galaxy10 parameter
    INPUT:
    OUTPUT:
        (instance): a callable instances from Cifar10_CNN with Galaxy10 parameter
    HISTORY:
        2018-Feb-09 - Written - Henry Leung (University of Toronto)
        2018-Apr-02 - Update - Henry Leung (University of Toronto)
    """
    from astroNN.datasets.galaxy10 import galaxy10cls_lookup
    from astroNN.models.misc_models import Cifar10CNN

    galaxy10_net = Cifar10CNN()
    galaxy10_net._model_identifier = "Galaxy10CNN"
    targetname = []
    for i in range(10):
        targetname.extend([galaxy10cls_lookup(i)])

    galaxy10_net.targetname = targetname
    return galaxy10_net


def convert_custom_objects(obj):
    """Handles custom object lookup.

    Based on Keras Source Code

    # Arguments
        obj: object, dict, or list.

    # Returns
        The same structure, where occurrences
            of a custom object name have been replaced
            with the custom object.
    """
    if isinstance(obj, list):
        deserialized = []
        for value in obj:
            deserialized.append(convert_custom_objects(value))
        return deserialized
    if isinstance(obj, dict):
        deserialized = {}
        for key, value in obj.items():
            deserialized[key] = convert_custom_objects(value)
        return deserialized
    return obj


[docs]def load_folder(folder=None): """ To load astroNN model object from folder :param folder: [optional] you should provide folder name if outside folder, do not specific when you are inside the folder :type folder: str :return: astroNN Neural Network instance :rtype: astroNN.nn.NeuralNetMaster.NeuralNetMaster :History: 2017-Dec-29 - Written - Henry Leung (University of Toronto) """ currentdir = os.getcwd() if folder is not None: fullfilepath = os.path.join(currentdir, folder) else: fullfilepath = currentdir astronn_model_obj = None if ( folder is not None and os.path.isfile(os.path.join(folder, "astroNN_model_parameter.json")) is True ): with open(os.path.join(folder, "astroNN_model_parameter.json")) as f: parameter = json.load(f) f.close() elif os.path.isfile("astroNN_model_parameter.json") is True: with open("astroNN_model_parameter.json") as f: parameter = json.load(f) f.close() elif folder is not None and not os.path.exists(folder): raise IOError("Folder not exists: " + str(currentdir + os.sep + folder)) else: raise FileNotFoundError("Are you sure this is an astroNN generated folder?") identifier = parameter["id"] unknown_model_message = f"Unknown model identifier -> {identifier}!" # need to point to the actual neural network if non-travial location if identifier == "Galaxy10CNN": astronn_model_obj = Galaxy10CNN() else: # else try to import it from standard way try: astronn_model_obj = getattr( importlib.import_module(f"astroNN.models"), identifier )() except ImportError: # try to load custom model from CUSTOM_MODEL_PATH if none are working CUSTOM_MODEL_PATH = custom_model_path_reader() # try the current folder and see if there is any .py on top of CUSTOM_MODEL_PATH list_py_files = [ os.path.join(fullfilepath, f) for f in os.listdir(fullfilepath) if f.endswith(".py") ] if CUSTOM_MODEL_PATH is None and list_py_files is None: print("\n") raise TypeError(unknown_model_message) else: for path_list in ( path_list for path_list in [CUSTOM_MODEL_PATH, list_py_files] if path_list is not None ): for path in path_list: head, tail = os.path.split(path) sys.path.insert(0, head) try: model = getattr( importlib.import_module(tail.strip(".py")), str(identifier), ) astronn_model_obj = model() except AttributeError: pass if astronn_model_obj is None: print("\n") raise TypeError(unknown_model_message) astronn_model_obj.currentdir = currentdir astronn_model_obj.fullfilepath = fullfilepath astronn_model_obj.folder_name = ( folder if folder is not None else os.path.basename(os.path.normpath(currentdir)) ) # Must have parameter astronn_model_obj._input_shape = parameter["input"] astronn_model_obj._labels_shape = parameter["labels"] if type(astronn_model_obj._input_shape) is not dict: astronn_model_obj._input_shape = {"input": astronn_model_obj._input_shape} if type(astronn_model_obj._labels_shape) is not dict: astronn_model_obj._labels_shape = {"output": astronn_model_obj._labels_shape} astronn_model_obj.num_hidden = parameter["hidden"] astronn_model_obj.input_norm_mode = parameter["input_norm_mode"] astronn_model_obj.labels_norm_mode = parameter["labels_norm_mode"] astronn_model_obj.batch_size = parameter["batch_size"] astronn_model_obj.targetname = parameter["targetname"] astronn_model_obj.val_size = parameter["valsize"] # Conditional parameter depends on neural net architecture try: astronn_model_obj.num_filters = parameter["filternum"] except KeyError: pass try: astronn_model_obj.filter_len = parameter["filterlen"] except KeyError: pass try: pool_length = parameter["pool_length"] if pool_length is not None: if isinstance(pool_length, int): # multi-dimensional case astronn_model_obj.pool_length = parameter["pool_length"] else: astronn_model_obj.pool_length = list(parameter["pool_length"]) except KeyError or TypeError: pass try: # need to convert to int because of keras do not want array or list astronn_model_obj.latent_dim = int(parameter["latent"]) except KeyError: pass try: astronn_model_obj.task = parameter["task"] except KeyError: pass try: astronn_model_obj.dropout_rate = parameter["dropout_rate"] except KeyError: pass try: # if inverse model precision exists, so does length_scale astronn_model_obj.inv_model_precision = parameter["inv_tau"] astronn_model_obj.length_scale = parameter["length_scale"] except KeyError: pass try: astronn_model_obj.l1 = parameter["l1"] except KeyError: pass try: astronn_model_obj.l2 = parameter["l2"] except KeyError: pass try: astronn_model_obj.maxnorm = parameter["maxnorm"] except KeyError: pass try: astronn_model_obj.input_names = parameter["input_names"] except KeyError: astronn_model_obj.input_names = ["input"] try: astronn_model_obj.output_names = parameter["output_names"] except KeyError: astronn_model_obj.output_names = ["output"] try: astronn_model_obj._last_layer_activation = parameter["last_layer_activation"] except KeyError: pass try: astronn_model_obj.activation = parameter["activation"] except KeyError: pass with h5py.File( os.path.join(astronn_model_obj.fullfilepath, "model_weights.h5"), mode="r" ) as f: training_config = json.loads(f.attrs["training_config"]) optimizer_config = training_config["optimizer_config"] optimizer = optimizers.deserialize(optimizer_config) model_config = json.loads(f.attrs["model_config"]) # input/name names, mean, std input_names = [] output_names = [] for lay in model_config["config"]["input_layers"]: input_names.append(lay[0]) for lay in model_config["config"]["output_layers"]: output_names.append(lay[0]) astronn_model_obj.input_mean = list_to_dict( input_names, dict_list_to_dict_np(parameter["input_mean"]) ) astronn_model_obj.labels_mean = list_to_dict( output_names, dict_list_to_dict_np(parameter["labels_mean"]) ) astronn_model_obj.input_std = list_to_dict( input_names, dict_list_to_dict_np(parameter["input_std"]) ) astronn_model_obj.labels_std = list_to_dict( output_names, dict_list_to_dict_np(parameter["labels_std"]) ) # Recover loss functions and metrics. losses_raw = convert_custom_objects(training_config["loss"]) if losses_raw: try: try: loss = [losses_lookup(losses_raw[_loss]) for _loss in losses_raw] except TypeError: loss = losses_lookup(losses_raw) except: raise LookupError("Cant lookup loss") else: loss=None metrics_raw = convert_custom_objects(training_config["metrics"]) # its weird that keras needs -> metrics[metric][0] instead of metrics[metric] likes losses try: try: if version.parse(tf.__version__) >= version.parse("2.4.0"): metrics = [ losses_lookup(_metric["config"]["fn"]) for _metric in metrics_raw[0] ] else: metrics = [ losses_lookup(metrics_raw[_metric][0]) for _metric in metrics_raw ] except TypeError: metrics = [losses_lookup(metrics_raw[0])] except: metrics = metrics_raw sample_weight_mode = ( training_config["sample_weight_mode"] if hasattr(training_config, "sample_weight_mode") else None ) loss_weights = training_config["loss_weights"] weighted_metrics = None # compile the model astronn_model_obj.compile( optimizer=optimizer, loss=loss, metrics=metrics, weighted_metrics=weighted_metrics, loss_weights=loss_weights, sample_weight_mode=sample_weight_mode, ) # set weights astronn_model_obj.keras_model.load_weights( os.path.join(astronn_model_obj.fullfilepath, "model_weights.h5") ) # Build train function (to get weight updates), need to consider Sequential model too astronn_model_obj.keras_model.make_train_function() try: optimizer_weights_group = f["optimizer_weights"] if version.parse(h5py.__version__) >= version.parse("3.0"): optimizer_weight_names = [ n for n in optimizer_weights_group.attrs["weight_names"] ] else: optimizer_weight_names = [ n.decode("utf8") for n in optimizer_weights_group.attrs["weight_names"] ] optimizer_weight_values = [ optimizer_weights_group[n] for n in optimizer_weight_names ] astronn_model_obj.keras_model.optimizer._create_all_weights( astronn_model_obj.keras_model.trainable_variables ) astronn_model_obj.keras_model.optimizer.set_weights(optimizer_weight_values) except KeyError: warnings.warn( "Error in loading the saved optimizer state. As a result, your model is starting with a freshly initialized optimizer." ) # create normalizer and set correct mean and std astronn_model_obj.input_normalizer = Normalizer( mode=astronn_model_obj.input_norm_mode ) astronn_model_obj.labels_normalizer = Normalizer( mode=astronn_model_obj.labels_norm_mode ) astronn_model_obj.input_normalizer.mean_labels = astronn_model_obj.input_mean astronn_model_obj.input_normalizer.std_labels = astronn_model_obj.input_std astronn_model_obj.labels_normalizer.mean_labels = astronn_model_obj.labels_mean astronn_model_obj.labels_normalizer.std_labels = astronn_model_obj.labels_std print("========================================================") print(f"Loaded astroNN model, model type: {astronn_model_obj.name} -> {identifier}") print("========================================================") return astronn_model_obj