# ---------------------------------------------------------#
# astroNN.models.apogee_models: Contain Apogee Models
# ---------------------------------------------------------#
import numpy as np
import tensorflow as tf
import tensorflow.keras as tfk
from astroNN.apogee import aspcap_mask
from astroNN.models.base_bayesian_cnn import BayesianCNNBase
from astroNN.models.base_cnn import CNNBase
from astroNN.models.base_vae import ConvVAEBase
from astroNN.nn.layers import MCDropout, BoolMask, StopGrad, KLDivergenceLayer, TensorInput
from astroNN.nn.losses import bayesian_binary_crossentropy_wrapper, bayesian_binary_crossentropy_var_wrapper
from astroNN.nn.losses import bayesian_categorical_crossentropy_wrapper, bayesian_categorical_crossentropy_var_wrapper
from astroNN.nn.losses import mse_lin_wrapper, mse_var_wrapper
Add = tfk.layers.Add
Dense = tfk.layers.Dense
Input = tfk.layers.Input
Conv1D = tfk.layers.Conv1D
Conv2D = tfk.layers.Conv2D
Lambda = tfk.layers.Lambda
Reshape = tfk.layers.Reshape
Dropout = tfk.layers.Dropout
Flatten = tfk.layers.Flatten
Multiply = tfk.layers.Multiply
Activation = tfk.layers.Activation
concatenate = tfk.layers.concatenate
MaxPooling1D = tfk.layers.MaxPooling1D
MaxPooling2D = tfk.layers.MaxPooling2D
Model = tfk.models.Model
Sequential = tfk.models.Sequential
regularizers = tfk.regularizers
MaxNorm = tfk.constraints.MaxNorm
RandomNormal = tfk.initializers.RandomNormal
# noinspection PyCallingNonCallable
[docs]class ApogeeBCNN(BayesianCNNBase):
"""
Class for Bayesian convolutional neural network for stellar spectra analysis
:History: 2017-Dec-21 - Written - Henry Leung (University of Toronto)
"""
def __init__(self, lr=0.0005, dropout_rate=0.3):
super().__init__()
self._implementation_version = '1.0'
self.initializer = RandomNormal(mean=0.0, stddev=0.05)
self.activation = 'relu'
self.num_filters = [2, 4]
self.filter_len = 8
self.pool_length = 4
self.num_hidden = [196, 96]
self.max_epochs = 100
self.lr = lr
self.reduce_lr_epsilon = 0.00005
self.reduce_lr_min = 1e-8
self.reduce_lr_patience = 2
self.l2 = 5e-9
self.dropout_rate = dropout_rate
self.input_norm_mode = 3
self.task = 'regression'
self.targetname = ['teff', 'logg', 'M', 'alpha', 'C', 'C1', 'N', 'O', 'Na', 'Mg', 'Al', 'Si', 'P', 'S', 'K',
'Ca', 'Ti', 'Ti2', 'V', 'Cr', 'Mn', 'Fe', 'Co', 'Ni', 'fakemag']
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
labels_err_tensor = Input(shape=(self._labels_shape['output'],), name='labels_err')
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(input_tensor)
activation_1 = Activation(activation=self.activation)(cnn_layer_1)
dropout_1 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_1)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(dropout_1)
activation_2 = Activation(activation=self.activation)(cnn_layer_2)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(activation_2)
flattener = Flatten()(maxpool_1)
dropout_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(flattener)
layer_3 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_2)
activation_3 = Activation(activation=self.activation)(layer_3)
dropout_3 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_3)
layer_4 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_3)
activation_4 = Activation(activation=self.activation)(layer_4)
output = Dense(units=self._labels_shape['output'], activation=self._last_layer_activation, name='output')(activation_4)
variance_output = Dense(units=self._labels_shape['output'], activation='linear', name='variance_output')(activation_4)
model = Model(inputs=[input_tensor, labels_err_tensor], outputs=[output, variance_output])
# new astroNN high performance dropout variational inference on GPU expects single output
model_prediction = Model(inputs=[input_tensor], outputs=concatenate([output, variance_output]))
if self.task == 'regression':
variance_loss = mse_var_wrapper(output, labels_err_tensor)
output_loss = mse_lin_wrapper(variance_output, labels_err_tensor)
elif self.task == 'classification':
output_loss = bayesian_categorical_crossentropy_wrapper(variance_output)
variance_loss = bayesian_categorical_crossentropy_var_wrapper(output)
elif self.task == 'binary_classification':
output_loss = bayesian_binary_crossentropy_wrapper(variance_output)
variance_loss = bayesian_binary_crossentropy_var_wrapper(output)
else:
raise RuntimeError('Only "regression", "classification" and "binary_classification" are supported')
return model, model_prediction, output_loss, variance_loss
# noinspection PyCallingNonCallable
[docs]class ApogeeBCNNCensored(BayesianCNNBase):
"""
Class for Bayesian censored convolutional neural network for stellar spectra analysis [specifically APOGEE
DR14 spectra only]
Described in the paper: https://ui.adsabs.harvard.edu/abs/2019MNRAS.483.3255L/abstract
:History: 2018-May-27 - Written - Henry Leung (University of Toronto)
"""
def __init__(self, lr=0.0005, dropout_rate=0.3):
super().__init__()
self._implementation_version = '1.0'
self.initializer = RandomNormal(mean=0.0, stddev=0.05)
self.activation = 'relu'
self.num_filters = [2, 4]
self.filter_len = 8
self.pool_length = 4
# number of neurone for [ApogeeBCNN_Dense_1, ApogeeBCNN_Dense_2, aspcap_1, aspcap_2, hidden]
self.num_hidden = [192, 96, 32, 16, 2]
self.max_epochs = 50
self.lr = lr
self.reduce_lr_epsilon = 0.00005
self.maxnorm = .5
self.reduce_lr_min = 1e-8
self.reduce_lr_patience = 2
self.l2 = 5e-9
self.dropout_rate = dropout_rate
self.input_norm_mode = 3
self.task = 'regression'
self.targetname = ['teff', 'logg', 'C', 'C1', 'N', 'O', 'Na', 'Mg', 'Al', 'Si', 'P', 'S', 'K', 'Ca', 'Ti',
'Ti2', 'V', 'Cr', 'Mn', 'Fe', 'Co', 'Ni']
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
input_tensor_flattened = Flatten()(input_tensor)
labels_err_tensor = Input(shape=(self._labels_shape['output'],), name='labels_err')
# slice spectra to censor out useless region for elements
censored_c_input = BoolMask(aspcap_mask("C", dr=14), name='C_Mask')(input_tensor_flattened)
censored_c1_input = BoolMask(aspcap_mask("C1", dr=14), name='C1_Mask')(input_tensor_flattened)
censored_n_input = BoolMask(aspcap_mask("N", dr=14), name='N_Mask')(input_tensor_flattened)
censored_o_input = BoolMask(aspcap_mask("O", dr=14), name='O_Mask')(input_tensor_flattened)
censored_na_input = BoolMask(aspcap_mask("Na", dr=14), name='Na_Mask')(input_tensor_flattened)
censored_mg_input = BoolMask(aspcap_mask("Mg", dr=14), name='Mg_Mask')(input_tensor_flattened)
censored_al_input = BoolMask(aspcap_mask("Al", dr=14), name='Al_Mask')(input_tensor_flattened)
censored_si_input = BoolMask(aspcap_mask("Si", dr=14), name='Si_Mask')(input_tensor_flattened)
censored_p_input = BoolMask(aspcap_mask("P", dr=14), name='P_Mask')(input_tensor_flattened)
censored_s_input = BoolMask(aspcap_mask("S", dr=14), name='S_Mask')(input_tensor_flattened)
censored_k_input = BoolMask(aspcap_mask("K", dr=14), name='K_Mask')(input_tensor_flattened)
censored_ca_input = BoolMask(aspcap_mask("Ca", dr=14), name='Ca_Mask')(input_tensor_flattened)
censored_ti_input = BoolMask(aspcap_mask("Ti", dr=14), name='Ti_Mask')(input_tensor_flattened)
censored_ti2_input = BoolMask(aspcap_mask("Ti2", dr=14), name='Ti2_Mask')(input_tensor_flattened)
censored_v_input = BoolMask(aspcap_mask("V", dr=14), name='V_Mask')(input_tensor_flattened)
censored_cr_input = BoolMask(aspcap_mask("Cr", dr=14), name='Cr_Mask')(input_tensor_flattened)
censored_mn_input = BoolMask(aspcap_mask("Mn", dr=14), name='Mn_Mask')(input_tensor_flattened)
censored_co_input = BoolMask(aspcap_mask("Co", dr=14), name='Co_Mask')(input_tensor_flattened)
censored_ni_input = BoolMask(aspcap_mask("Ni", dr=14), name='Ni_Mask')(input_tensor_flattened)
# get neurones from each elements from censored spectra
c_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2] * 8, kernel_initializer=self.initializer, name='c_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_c_input))
c1_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='c1_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_c1_input))
n_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2] * 8, kernel_initializer=self.initializer, name='n_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_n_input))
o_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='o_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_o_input))
na_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='na_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_na_input))
mg_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='mg_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_mg_input))
al_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='al_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_al_input))
si_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='si_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_si_input))
p_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='p_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_p_input))
s_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='s_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_s_input))
k_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='k_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_k_input))
ca_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='ca_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_ca_input))
ti_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='ti_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_ti_input))
ti2_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='ti2_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_ti2_input))
v_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='v_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_v_input))
cr_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='cr_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_cr_input))
mn_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='mn_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_mn_input))
co_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='co_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_co_input))
ni_dense = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[2], kernel_initializer=self.initializer, name='ni_dense',
activation=self.activation, kernel_regularizer=regularizers.l2(self.l2))(censored_ni_input))
# get neurones from each elements from censored spectra
c_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3] * 4, kernel_initializer=self.initializer, activation=self.activation,
name='c_dense_2')(c_dense))
c1_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='c1_dense_2')(c1_dense))
n_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3] * 4, kernel_initializer=self.initializer, activation=self.activation,
name='n_dense_2')(n_dense))
o_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='o_dense_2')(o_dense))
na_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='na_dense_2')(na_dense))
mg_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='mg_dense_2')(mg_dense))
al_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='al_dense_2')(al_dense))
si_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='si_dense_2')(si_dense))
p_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='p_dense_2')(p_dense))
s_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='s_dense_2')(s_dense))
k_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='k_dense_2')(k_dense))
ca_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='ca_dense_2')(ca_dense))
ti_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='ti_dense_2')(ti_dense))
ti2_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='ti2_dense_2')(ti2_dense))
v_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='v_dense_2')(v_dense))
cr_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='cr_dense_2')(cr_dense))
mn_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='mn_dense_2')(mn_dense))
co_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='co_dense_2')(co_dense))
ni_dense_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(
Dense(units=self.num_hidden[3], kernel_initializer=self.initializer, activation=self.activation,
name='ni_dense_2')(ni_dense))
# Basically the same as ApogeeBCNN structure
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(input_tensor)
activation_1 = Activation(activation=self.activation)(cnn_layer_1)
dropout_1 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_1)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(dropout_1)
activation_2 = Activation(activation=self.activation)(cnn_layer_2)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(activation_2)
flattener = Flatten()(maxpool_1)
dropout_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(flattener)
layer_3 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_2)
activation_3 = Activation(activation=self.activation)(layer_3)
dropout_3 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_3)
layer_4 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_3)
activation_4 = Activation(activation=self.activation)(layer_4)
teff_output = Dense(units=1)(activation_4)
logg_output = Dense(units=1)(activation_4)
fe_output = Dense(units=1)(activation_4)
old_3_output_wo_grad = StopGrad()(concatenate([teff_output, logg_output, fe_output]))
teff_output_var = Dense(units=1)(activation_4)
logg_output_var = Dense(units=1)(activation_4)
fe_output_var = Dense(units=1)(activation_4)
aux_fullspec = Dense(units=self.num_hidden[4], kernel_initializer=self.initializer,
kernel_constraint=MaxNorm(self.maxnorm), name='aux_fullspec')(activation_4)
fullspec_hidden = concatenate([aux_fullspec, old_3_output_wo_grad])
# get the final answer
c_concat = Dense(units=1, name='c_concat')(concatenate([c_dense_2, fullspec_hidden]))
c1_concat = Dense(units=1, name='c1_concat')(concatenate([c1_dense_2, fullspec_hidden]))
n_concat = Dense(units=1, name='n_concat')(concatenate([n_dense_2, fullspec_hidden]))
o_concat = Dense(units=1, name='o_concat')(concatenate([o_dense_2, fullspec_hidden]))
na_concat = Dense(units=1, name='na_concat')(concatenate([na_dense_2, fullspec_hidden]))
mg_concat = Dense(units=1, name='mg_concat')(concatenate([mg_dense_2, fullspec_hidden]))
al_concat = Dense(units=1, name='al_concat')(concatenate([al_dense_2, fullspec_hidden]))
si_concat = Dense(units=1, name='si_concat')(concatenate([si_dense_2, fullspec_hidden]))
p_concat = Dense(units=1, name='p_concat')(concatenate([p_dense_2, fullspec_hidden]))
s_concat = Dense(units=1, name='s_concat')(concatenate([s_dense_2, fullspec_hidden]))
k_concat = Dense(units=1, name='k_concat')(concatenate([k_dense_2, fullspec_hidden]))
ca_concat = Dense(units=1, name='ca_concat')(concatenate([ca_dense_2, fullspec_hidden]))
ti_concat = Dense(units=1, name='ti_concat')(concatenate([ti_dense_2, fullspec_hidden]))
ti2_concat = Dense(units=1, name='ti2_concat')(concatenate([ti2_dense_2, fullspec_hidden]))
v_concat = Dense(units=1, name='v_concat')(concatenate([v_dense_2, fullspec_hidden]))
cr_concat = Dense(units=1, name='cr_concat')(concatenate([cr_dense_2, fullspec_hidden]))
mn_concat = Dense(units=1, name='mn_concat')(concatenate([mn_dense_2, fullspec_hidden]))
co_concat = Dense(units=1, name='co_concat')(concatenate([co_dense_2, fullspec_hidden]))
ni_concat = Dense(units=1, name='ni_concat')(concatenate([ni_dense_2, fullspec_hidden]))
# get the final predictive uncertainty
c_concat_var = Dense(units=1, name='c_concat_var')(concatenate([c_dense_2, fullspec_hidden]))
c1_concat_var = Dense(units=1, name='c1_concat_var')(concatenate([c1_dense_2, fullspec_hidden]))
n_concat_var = Dense(units=1, name='n_concat_var')(concatenate([n_dense_2, fullspec_hidden]))
o_concat_var = Dense(units=1, name='o_concat_var')(concatenate([o_dense_2, fullspec_hidden]))
na_concat_var = Dense(units=1, name='na_concat_var')(concatenate([na_dense_2, fullspec_hidden]))
mg_concat_var = Dense(units=1, name='mg_concat_var')(concatenate([mg_dense_2, fullspec_hidden]))
al_concat_var = Dense(units=1, name='al_concat_var')(concatenate([al_dense_2, fullspec_hidden]))
si_concat_var = Dense(units=1, name='si_concat_var')(concatenate([si_dense_2, fullspec_hidden]))
p_concat_var = Dense(units=1, name='p_concat_var')(concatenate([p_dense_2, fullspec_hidden]))
s_concat_var = Dense(units=1, name='s_concat_var')(concatenate([s_dense_2, fullspec_hidden]))
k_concat_var = Dense(units=1, name='k_concat_var')(concatenate([k_dense_2, fullspec_hidden]))
ca_concat_var = Dense(units=1, name='ca_concat_var')(concatenate([ca_dense_2, fullspec_hidden]))
ti_concat_var = Dense(units=1, name='ti_concat_var')(concatenate([ti_dense_2, fullspec_hidden]))
ti2_concat_var = Dense(units=1, name='ti2_concat_var')(concatenate([ti2_dense_2, fullspec_hidden]))
v_concat_var = Dense(units=1, name='v_concat_var')(concatenate([v_dense_2, fullspec_hidden]))
cr_concat_var = Dense(units=1, name='cr_concat_var')(concatenate([cr_dense_2, fullspec_hidden]))
mn_concat_var = Dense(units=1, name='mn_concat_var')(concatenate([mn_dense_2, fullspec_hidden]))
co_concat_var = Dense(units=1, name='co_concat_var')(concatenate([co_dense_2, fullspec_hidden]))
ni_concat_var = Dense(units=1, name='ni_concat_var')(concatenate([ni_dense_2, fullspec_hidden]))
# concatenate answer
output = concatenate([teff_output, logg_output, c_concat, c1_concat, n_concat, o_concat, na_concat, mg_concat,
al_concat, si_concat, p_concat, s_concat, k_concat, ca_concat, ti_concat, ti2_concat,
v_concat, cr_concat, mn_concat, fe_output, co_concat, ni_concat], name='output')
# concatenate predictive uncertainty
variance_output = concatenate([teff_output_var, logg_output_var, c_concat_var, c1_concat_var, n_concat_var,
o_concat_var, na_concat_var, mg_concat_var, al_concat_var, si_concat_var,
p_concat_var, s_concat_var, k_concat_var, ca_concat_var, ti_concat_var,
ti2_concat_var, v_concat_var, cr_concat_var, mn_concat_var, fe_output_var,
co_concat_var, ni_concat_var], name='variance_output')
model = Model(inputs=[input_tensor, labels_err_tensor], outputs=[output, variance_output])
# new astroNN high performance dropout variational inference on GPU expects single output
model_prediction = Model(inputs=input_tensor, outputs=concatenate([output, variance_output]))
variance_loss = mse_var_wrapper(output, labels_err_tensor)
output_loss = mse_lin_wrapper(variance_output, labels_err_tensor)
return model, model_prediction, output_loss, variance_loss
[docs]class ApogeeCNN(CNNBase):
"""
Class for Convolutional Neural Network for stellar spectra analysis
:History: 2017-Dec-21 - Written - Henry Leung (University of Toronto)
"""
def __init__(self, lr=0.005):
super().__init__()
self._implementation_version = '1.0'
self.initializer = 'he_normal'
self.activation = 'relu'
self.num_filters = [2, 4]
self.filter_len = 8
self.pool_length = 4
self.num_hidden = [196, 96]
self.max_epochs = 100
self.lr = lr
self.reduce_lr_epsilon = 0.00005
self.reduce_lr_min = 1e-8
self.reduce_lr_patience = 2
self.l2 = 1e-5
self.dropout_rate = 0.1
self.input_norm_mode = 3
self.task = 'regression'
self.targetname = ['teff', 'logg', 'M', 'alpha', 'C', 'C1', 'N', 'O', 'Na', 'Mg', 'Al', 'Si', 'P', 'S', 'K',
'Ca', 'Ti', 'Ti2', 'V', 'Cr', 'Mn', 'Fe', 'Co', 'Ni', 'fakemag']
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(input_tensor)
activation_1 = Activation(activation=self.activation)(cnn_layer_1)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(activation_1)
activation_2 = Activation(activation=self.activation)(cnn_layer_2)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(activation_2)
flattener = Flatten()(maxpool_1)
dropout_1 = Dropout(self.dropout_rate)(flattener)
layer_3 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_1)
activation_3 = Activation(activation=self.activation)(layer_3)
dropout_2 = Dropout(self.dropout_rate)(activation_3)
layer_4 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_2)
activation_4 = Activation(activation=self.activation)(layer_4)
layer_5 = Dense(units=self._labels_shape['output'])(activation_4)
output = Activation(activation=self._last_layer_activation, name='output')(layer_5)
model = Model(inputs=input_tensor, outputs=output)
return model
[docs]class StarNet2017(CNNBase):
"""
To create StarNet, S. Fabbro et al. (2017) arXiv:1709.09182. astroNN implemented the exact architecture with
default parameter same as StarNet paper
:History: 2017-Dec-23 - Written - Henry Leung (University of Toronto)
"""
def __init__(self):
super().__init__()
self.name = 'StarNet (arXiv:1709.09182)'
self._implementation_version = '1.0'
self.initializer = 'he_normal'
self.activation = 'relu'
self.num_filters = [4, 16]
self.filter_len = 8
self.pool_length = 4
self.num_hidden = [256, 128]
self.max_epochs = 30
self.lr = 0.0007
self.l2 = 0.
self.reduce_lr_epsilon = 0.00005
self.reduce_lr_min = 0.00008
self.reduce_lr_patience = 2
self.early_stopping_min_delta = 0.0001
self.early_stopping_patience = 4
self.input_norm_mode = 3
self.task = 'regression'
self.targetname = ['teff', 'logg', 'Fe']
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[0], kernel_size=self.filter_len)(input_tensor)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[1], kernel_size=self.filter_len)(cnn_layer_1)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(cnn_layer_2)
flattener = Flatten()(maxpool_1)
layer_3 = Dense(units=self.num_hidden[0], kernel_initializer=self.initializer, activation=self.activation)(
flattener)
layer_4 = Dense(units=self.num_hidden[1], kernel_initializer=self.initializer, activation=self.activation)(
layer_3)
layer_out = Dense(units=self._labels_shape['output'], kernel_initializer=self.initializer,
activation=self._last_layer_activation, name='output')(layer_4)
model = Model(inputs=input_tensor, outputs=layer_out)
return model
# noinspection PyCallingNonCallable
[docs]class ApogeeCVAE(ConvVAEBase):
"""
Class for Convolutional Autoencoder Neural Network for stellar spectra analysis
:History: 2017-Dec-21 - Written - Henry Leung (University of Toronto)
"""
def __init__(self):
super().__init__()
self._implementation_version = '1.0'
self.batch_size = 64
self.initializer = 'he_normal'
self.activation = 'relu'
self.optimizer = None
self.num_filters = [2, 4]
self.filter_len = 8
self.pool_length = 4
self.num_hidden = [128, 64]
self.latent_dim = 2
self.max_epochs = 100
self.lr = 0.0005
self.reduce_lr_epsilon = 0.0005
self.reduce_lr_min = 0.0000000001
self.reduce_lr_patience = 4
self.epsilon_std = 1.0
self.task = 'regression'
self.keras_encoder = None
self.keras_vae = None
self.l1 = 1e-5
self.l2 = 1e-5
self.dropout_rate = 0.1
self._last_layer_activation = 'linear'
self.targetname = 'Spectra Reconstruction'
self.input_norm_mode = '2'
self.labels_norm_mode = '2'
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(input_tensor)
dropout_1 = Dropout(self.dropout_rate)(cnn_layer_1)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(dropout_1)
dropout_2 = Dropout(self.dropout_rate)(cnn_layer_2)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(dropout_2)
flattener = Flatten()(maxpool_1)
layer_4 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l1(self.l1),
kernel_initializer=self.initializer, activation=self.activation)(flattener)
dropout_3 = Dropout(self.dropout_rate)(layer_4)
layer_5 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l1(self.l1),
kernel_initializer=self.initializer, activation=self.activation)(dropout_3)
dropout_4 = Dropout(self.dropout_rate)(layer_5)
z_mu = Dense(units=self.latent_dim, activation="linear", name='mean_output',
kernel_initializer=self.initializer,
kernel_regularizer=regularizers.l1(self.l1))(dropout_4)
z_log_var = Dense(units=self.latent_dim, activation='linear', name='sigma_output',
kernel_initializer=self.initializer,
kernel_regularizer=regularizers.l1(self.l1))(dropout_4)
z_mu, z_log_var = KLDivergenceLayer()([z_mu, z_log_var])
z_sigma = Lambda(lambda t: tf.exp(.5 * t))(z_log_var)
z_eps = Multiply()([z_sigma, tf.random.normal(mean=0., stddev=self.epsilon_std, shape=(tf.shape(z_mu)[0], self.latent_dim))])
z = Add()([z_mu, z_eps])
decoder = Sequential(name='output')
decoder.add(Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l1(self.l1),
kernel_initializer=self.initializer, activation=self.activation, input_dim=self.latent_dim))
decoder.add(Dropout(self.dropout_rate))
decoder.add(Dense(units=self._input_shape['input'][0] * self.num_filters[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer, activation=self.activation))
decoder.add(Dropout(self.dropout_rate))
output_shape = (self.batch_size, self._input_shape['input'][0], self.num_filters[1])
decoder.add(Reshape(output_shape[1:]))
decoder.add(Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2)))
decoder.add(Dropout(self.dropout_rate))
decoder.add(Conv1D(kernel_initializer=self.initializer, activation=self.activation, padding="same",
filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2)))
decoder.add(Conv1D(kernel_initializer=self.initializer, activation=self._last_layer_activation, padding="same",
filters=1, kernel_size=self.filter_len, name='output'))
x_pred = decoder(z)
vae = Model(inputs=[input_tensor], outputs=[x_pred])
encoder = Model(inputs=[input_tensor], outputs=[z_mu])
return vae, encoder, decoder
class DeNormAdd(tfk.layers.Layer):
"""
Just a layer to work around `TypeError: can't pickle _thread.lock objects` issue when saving this particular model
For denormalizing
"""
def __init__(self, norm, name=None, **kwargs):
self.norm = norm
self.supports_masking = True
if not name:
prefix = self.__class__.__name__
name = prefix + '_' + str(tfk.backend.get_uid(prefix))
super().__init__(name=name, **kwargs)
def call(self, inputs, training=None):
return tf.add(inputs, self.norm)
def get_config(self):
"""
:return: Dictionary of configuration
:rtype: dict
"""
config = {'norm': self.norm}
base_config = super().get_config()
return {**dict(base_config.items()), **config}
# noinspection PyCallingNonCallable
[docs]class ApogeeDR14GaiaDR2BCNN(BayesianCNNBase):
"""
Class for Bayesian convolutional neural network for APOGEE DR14 Gaia DR2
:History: 2018-Nov-06 - Written - Henry Leung (University of Toronto)
"""
def __init__(self, lr=0.001, dropout_rate=0.3):
super().__init__()
self._implementation_version = '1.0'
self.initializer = RandomNormal(mean=0.0, stddev=0.05)
self.activation = 'relu'
self.num_filters = [2, 4]
self.filter_len = 8
self.pool_length = 4
self.num_hidden = [162, 64, 32, 16]
self.max_epochs = 100
self.lr = lr
self.reduce_lr_epsilon = 0.00005
self.reduce_lr_min = 1e-8
self.reduce_lr_patience = 2
self.l2 = 5e-9
self.dropout_rate = dropout_rate
self.input_norm_mode = 3
self.task = 'regression'
self.targetname = ['Ks-band fakemag']
def magmask(self):
magmask = np.zeros(self._input_shape['input'][0], dtype=np.bool)
magmask[7514] = True # mask to extract extinction correction apparent magnitude
return magmask
def specmask(self):
specmask = np.zeros(self._input_shape['input'][0], dtype=np.bool)
specmask[:7514] = True # mask to extract extinction correction apparent magnitude
return specmask
def gaia_aux_mask(self):
gaia_aux = np.zeros(self._input_shape['input'][0], dtype=np.bool)
gaia_aux[7515:] = True # mask to extract data for gaia offset
return gaia_aux
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input') # training data
labels_err_tensor = Input(shape=(self._labels_shape['output'],), name='labels_err')
# extract spectra from input data and expand_dims for convolution
spectra = Lambda(lambda x: tf.expand_dims(x, axis=-1))(BoolMask(self.specmask())(Flatten()(input_tensor)))
# value to denorm magnitude
app_mag = BoolMask(self.magmask())(Flatten()(input_tensor))
# tf.convert_to_tensor(self.input_mean[self.magmask()])
denorm_mag = DeNormAdd(np.array(self.input_mean['input'][self.magmask()]))(app_mag)
inv_pow_mag = Lambda(lambda mag: tf.pow(10., tf.multiply(-0.2, mag)))(denorm_mag)
# data to infer Gia DR2 offset
# ========================== Offset Calibration Model ========================== #
gaia_aux_data = BoolMask(self.gaia_aux_mask())(Flatten()(input_tensor))
gaia_aux_hidden = MCDropout(self.dropout_rate, disable=self.disable_dropout)(Dense(units=self.num_hidden[2],
kernel_regularizer=regularizers.l2(
self.l2),
kernel_initializer=self.initializer,
activation='tanh')(
gaia_aux_data))
gaia_aux_hidden2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(Dense(units=self.num_hidden[3],
kernel_regularizer=regularizers.l2(
self.l2),
kernel_initializer=self.initializer,
activation='tanh')(
gaia_aux_hidden))
offset = Dense(units=1, kernel_initializer=self.initializer, activation='tanh', name='offset_output')(
gaia_aux_hidden2)
# ========================== Offset Calibration Model ========================== #
# good old NN takes spectra and output fakemag
# ========================== Spectro-Luminosity Model ========================== #
cnn_layer_1 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[0],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(spectra)
activation_1 = Activation(activation=self.activation)(cnn_layer_1)
dropout_1 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_1)
cnn_layer_2 = Conv1D(kernel_initializer=self.initializer, padding="same", filters=self.num_filters[1],
kernel_size=self.filter_len, kernel_regularizer=regularizers.l2(self.l2))(dropout_1)
activation_2 = Activation(activation=self.activation)(cnn_layer_2)
maxpool_1 = MaxPooling1D(pool_size=self.pool_length)(activation_2)
flattener = Flatten()(maxpool_1)
dropout_2 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(flattener)
layer_3 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_2)
activation_3 = Activation(activation=self.activation)(layer_3)
dropout_3 = MCDropout(self.dropout_rate, disable=self.disable_dropout)(activation_3)
layer_4 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_3)
activation_4 = Activation(activation=self.activation)(layer_4)
fakemag_output = Dense(units=self._labels_shape['output'], activation='softplus', name='fakemag_output')(activation_4)
fakemag_variance_output = Dense(units=self._labels_shape['output'], activation='linear',
name='fakemag_variance_output')(activation_4)
# ========================== Spectro-Luminosity Model ========================== #
# multiply a pre-determined de-normalization factor, such that fakemag std approx. 1 for training set
# it does not really matter as NN will adapt to whatever value this is
_fakemag_denorm = Lambda(lambda x: tf.multiply(x, 73.85))(fakemag_output)
_fakemag_var_denorm = Lambda(lambda x: tf.add(x, tf.math.log(73.85)))(fakemag_variance_output)
_fakemag_parallax = Multiply()([_fakemag_denorm, inv_pow_mag])
# output parallax
output = Add(name='output')([_fakemag_parallax, offset])
variance_output = Lambda(lambda x: tf.math.log(tf.abs(tf.multiply(x[2], tf.divide(tf.exp(x[0]), x[1])))),
name='variance_output')([fakemag_variance_output, fakemag_output, _fakemag_parallax])
model = Model(inputs=[input_tensor, labels_err_tensor], outputs=[output, variance_output])
# new astroNN high performance dropout variational inference on GPU expects single output
# while training with parallax, we want testing output fakemag
model_prediction = Model(inputs=[input_tensor], outputs=concatenate([_fakemag_denorm, _fakemag_var_denorm]))
variance_loss = mse_var_wrapper(output, labels_err_tensor)
output_loss = mse_lin_wrapper(variance_output, labels_err_tensor)
return model, model_prediction, output_loss, variance_loss
class ApogeeKplerEchelle(CNNBase):
"""
Class for Convolutional Neural Network for Echelle Diagram
:History: 2020-Apr-06 - Written - Henry Leung (University of Toronto)
"""
def __init__(self, lr=0.002):
super().__init__()
self._implementation_version = '1.0'
self.initializer = 'glorot_uniform'
self.activation = 'tanh'
self.num_filters = [2, 4]
self.filter_len = [8, 8]
self.pool_length = 4
self.num_hidden = [64, 32]
self.max_epochs = 40
self.lr = lr
self.reduce_lr_epsilon = 0.00005
self.reduce_lr_min = 1e-8
self.reduce_lr_patience = 2
self.l2 = 0.
self.dropout_rate = 0.1
self.input_norm_mode = {'input': 255, 'aux': 2}
self.labels_norm_mode = 2
self.task = 'regression'
self.targetname = []
def model(self):
input_tensor = Input(shape=self._input_shape['input'], name='input')
aux_tensor = Input(shape=self._input_shape['aux'], name='aux')
aux_flatten = Flatten()(aux_tensor)
cnn_layer_1 = Conv2D(kernel_initializer=self.initializer, padding="valid",
filters=self.num_filters[0], kernel_size=self.filter_len)(input_tensor)
activation_1 = Activation(activation=self.activation)(cnn_layer_1)
cnn_layer_2 = Conv2D(kernel_initializer=self.initializer, padding="valid",
filters=self.num_filters[1], kernel_size=self.filter_len)(activation_1)
activation_2 = Activation(activation=self.activation)(cnn_layer_2)
maxpool_1 = MaxPooling2D(pool_size=self.pool_length)(activation_2)
flattener = Flatten()(maxpool_1)
dropout_1 = Dropout(self.dropout_rate)(flattener)
layer_3 = Dense(units=self.num_hidden[0], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(dropout_1)
activation_3 = Activation(activation=self.activation)(layer_3)
dropout_2 = Dropout(self.dropout_rate)(activation_3)
layer_4 = Dense(units=self.num_hidden[1], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(concatenate([dropout_2, aux_flatten]))
activation_4 = Activation(activation=self.activation)(layer_4)
layer_5 = Dense(units=self._labels_shape['output'], kernel_regularizer=regularizers.l2(self.l2),
kernel_initializer=self.initializer)(activation_4)
output = Activation(activation=self._last_layer_activation, name='output')(layer_5)
model = Model(inputs=[input_tensor, aux_tensor], outputs=[output])
return model
