У меня есть модель, которая обрабатывает 2 входных 3D-изображения одинакового размера, A и B, для использования в более классической функции, чтобы попытаться повысить производительность этой функции. Чтобы правильно обучить модель, мне нужно применить функцию к результату каждого прогона. Сама функция принимает 2 значения, которые соответствуют значениям в A и B с одинаковыми координатами p. Этот результат должен быть сохранен в трехмерном изображении C того же размера, что и A и B в точке p. Классические реализации этого будут выполнять for l oop по всем координатам и применять функцию для каждой пары. К сожалению, этот подход не работает для обучения модели Keras, так как выходные данные функции должны возвращаться к весам предыдущих слоев.
Input -(A, B)-> Model -(A', B')-> Function(A'[p], B'[p]) -(C[p])-> Result
Я попытался написать собственный слой Keras для этого , Этот слой принимает 4D-тензор (channel, z, y, x) и должен возвращать тензор с формой (1, z, y, x).
В настоящее время это выглядит следующим образом:
# imports
def function(x: [float, float]) -> float:
# a -> x[0], b -> x[1]
# Calculate
return c
class CustomLayer(Layer):
# ... __init__ and build
def call(self, inputs, **kwargs):
# All samples, channel n ([::][n])
# We stack the tensors in such a way because map_fn() maps the top most axis to the function.
# This way a tensor of shape (n_voxels, 2) is created and the values are delivered in pairs to the function
map_input = K.stack([K.flatten(inputs[::][0]), K.flatten(inputs[::][1]), axis=1])
result = K.map_fn(lambda x: function(x), map_input)
result = K.reshape(result, K.constant([-1, 1, inputs.shape[2], inputs.shape[3], inputs.shape[4]], dtype=tf.int32))
return result
К сожалению, этот метод серьезно замедлил обучение , В то время как модели без пользовательского слоя в конце потребовалось около 45 минут для обучения за эпоху, модель с пользовательским слоем занимает около 120 часов за эпоху.
Модель и функция сами по себе могут выполнить требуемую задачу со значительной ошибкой, однако я хотел посмотреть, смогу ли я объединить их для получения лучших результатов.
Реальное Пожизненным использованием в моем случае является разложение материалов из
Dual Energy CT . Вы можете рассчитать долю материалов для каждого вокселя, предположив 3 известных материала
mat1,
mat2,
mat3 и неизвестный образец
sample.
. С помощью некоторых вычислений вы можете затем разложить этот неизвестный образец. на доли каждого известного материала f1, f2, f3 (f1 + f2 + f3 == 1,0). Нас интересует только f3, поэтому расчеты для других дробей были опущены.
Фактический код:
"""
DECT Decomposition Layer
"""
import numpy as np
import tensorflow as tf
from keras import backend as K
from keras.layers import Layer
from keras.constraints import MinMaxNorm
def _intersect(line1_start, line1_end, line2_start, line2_end, ):
"""
Find the intersection point between 2 lines
"""
# https://en.wikipedia.org/wiki/Line%E2%80%93line_intersection
# Tensorflow's tensors need a little more lines to unpack
x1 = line1_start[0]
y1 = line1_start[1]
x2 = line1_end[0]
y2 = line1_end[1]
x3 = line2_start[0]
y3 = line2_start[1]
x4 = line2_end[0]
y4 = line2_end[1]
px = (x1 * y2 - y1 * x2) * (x3 - x4) - (x1 - x2) * (x3 * y4 - y3 * x4)
px /= (x1 - x2) * (y3 - y4) - (y1 - y2) * (x3 - x4)
py = (x1 * y2 - y1 * x2) * (y3 - y4) - (y1 - y2) * (x3 * y4 - y3 * x4)
py /= (x1 - x2) * (y3 - y4) - (y1 - y2) * (x3 - x4)
return K.stack([px, py])
def _decomp(sample, mat1, mat2, mat3):
"""
Decomposition of a sample into 1 fraction of material 3
"""
# Calculate the sample lines' ends
sample3 = sample + (mat2 - mat3)
# Calculate the intersection points between the sample lines and triangle sides
intersect3 = _intersect(sample, sample3, mat1, mat2)
# Find out how far along the sample line the intersection is
f3 = tf.norm(sample - intersect3) / tf.norm(sample - sample3)
return f3
class DectDecompoLayer(Layer):
def __init__(self, mat1, mat2, mat3, **kwargs):
self.mat1 = K.constant(mat1)
self.mat2 = K.constant(mat2)
self.mat3 = K.constant(mat3)
super(DectDecompoLayer, self).__init__(**kwargs)
def build(self, input_shape):
super(DectDecompoLayer, self).build(input_shape)
def call(self, inputs, **kwargs):
map_input = K.stack([K.flatten(inputs[::][0]), K.flatten(inputs[::][1])], axis=1)
result = K.map_fn(lambda x: _decomp(x, self.mat1, self.mat2, self.mat3), map_input)
result = K.reshape(result, K.constant([-1, 1, inputs.shape[2], inputs.shape[3], inputs.shape[4]], dtype=tf.int32))
return result
def compute_output_shape(self, input_shape):
return input_shape[0], 1, input_shape[2], input_shape[3], input_shape[4]