from collections import namedtuple from keras.models import Model from . import networks from .metrics import * NetworkParameters = namedtuple("NetworkParameters", [ "type", "flow_features", "window_size", "domain_length", "output", "embedding_size", "domain_filter", "domain_kernel", "domain_dense", "domain_dropout", "main_filter", "main_kernel", "main_dense", "main_dropout", ]) def create_model(model, output_type): if output_type == "both": return Model(inputs=[model.in_domains, model.in_flows], outputs=(model.out_client, model.out_server)) elif output_type == "client": return Model(inputs=[model.in_domains, model.in_flows], outputs=(model.out_client,)) else: raise Exception("unknown model output") def get_models_by_params(params: dict): K.clear_session() # decomposing param section # mainly embedding model embedding_type = params.get("embedding_type", "small") network_type = params.get("type") # network_depth = params.get("depth") embedding_size = params.get("embedding") filter_embedding = params.get("filter_embedding") kernel_embedding = params.get("kernel_embedding") hidden_embedding = params.get("dense_embedding") # dropout = params.get("dropout") # mainly prediction model flow_features = params.get("flow_features") window_size = params.get("window_size") domain_length = params.get("domain_length") filter_main = params.get("filter_main") kernel_main = params.get("kernel_main") dense_dim = params.get("dense_main") model_output = params.get("model_output", "both") if embedding_type == "small": domain_cnn = networks.get_domain_embedding_model(embedding_size, domain_length, filter_embedding, kernel_embedding, hidden_embedding, 0.5) elif embedding_type == "deep": domain_cnn = networks.get_domain_embedding_model2(embedding_size, domain_length, filter_embedding, kernel_embedding, hidden_embedding, 0.5) else: raise ValueError("embedding type not found") if network_type == "final": model = networks.get_final_model(0.25, flow_features, window_size, domain_length, filter_main, kernel_main, dense_dim, domain_cnn) model = create_model(model, model_output) elif network_type in ("inter", "staggered"): model = networks.get_inter_model(0.25, flow_features, window_size, domain_length, filter_main, kernel_main, dense_dim, domain_cnn) model = create_model(model, model_output) elif network_type == "long": model = networks.get_long_model(0.25, flow_features, window_size, domain_length, filter_main, kernel_main, dense_dim, domain_cnn) model = create_model(model, model_output) elif network_type == "soft": model = networks.get_long_model(0.25, flow_features, window_size, domain_length, filter_main, kernel_main, dense_dim, domain_cnn) model = create_model(model, model_output) conv_server = model.get_layer("conv_server").trainable_weights conv_client = model.get_layer("conv_client").trainable_weights l1 = [0.001 * K.sum(K.abs(x - y)) for (x, y) in zip(conv_server, conv_client)] model.add_loss(l1) dense_server = model.get_layer("dense_server").trainable_weights dense_client = model.get_layer("dense_client").trainable_weights l2 = [0.001 * K.sum(K.abs(x - y)) for (x, y) in zip(dense_server, dense_client)] model.add_loss(l2) elif network_type == "sluice": model = networks.get_sluice_model(0.25, flow_features, window_size, domain_length, filter_main, kernel_main, dense_dim, domain_cnn) model = create_model(model, model_output) conv_server = model.get_layer("conv_server").trainable_weights conv_client = model.get_layer("conv_client").trainable_weights l1 = [0.001 * K.sum(K.abs(x - y)) for (x, y) in zip(conv_server, conv_client)] model.add_loss(l1) dense_server = model.get_layer("dense_server").trainable_weights dense_client = model.get_layer("dense_client").trainable_weights l2 = [0.001 * K.sum(K.abs(x - y)) for (x, y) in zip(dense_server, dense_client)] model.add_loss(l2) else: raise ValueError("network type not found") return model def get_server_model_by_params(params: dict): # decomposing param section # mainly embedding model network_depth = params.get("depth") embedding_size = params.get("embedding") input_length = params.get("input_length") filter_embedding = params.get("filter_embedding") kernel_embedding = params.get("kernel_embedding") hidden_embedding = params.get("dense_embedding") # mainly prediction model flow_features = params.get("flow_features") domain_length = params.get("domain_length") dense_dim = params.get("dense_main") embedding_model = networks.get_domain_embedding_model(embedding_size, input_length, filter_embedding, kernel_embedding, hidden_embedding, 0.5) return networks.get_server_model(flow_features, domain_length, dense_dim, embedding_model)