import gradio as gr import os import matplotlib.pyplot as plt from scipy.integrate import odeint import torch from torch.utils import data from torch.utils.data import DataLoader, Dataset from torch import nn, optim from skimage.transform import rescale, resize from torch import nn, optim import torch.nn.functional as F from torch.utils.data import Subset from scipy.interpolate import interp1d import collections import numpy as np import random #for pvloop simulator: import pandas as pd from scipy.integrate import odeint import torchvision import echonet import matplotlib.animation as animation from functools import partial device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") sequences_all = [] info_data_all = [] path = 'EchoNet-Dynamic' output_path = '' class Echo(torchvision.datasets.VisionDataset): """EchoNet-Dynamic Dataset. Args: root (string): Root directory of dataset (defaults to `echonet.config.DATA_DIR`) split (string): One of {``train'', ``val'', ``test'', ``all'', or ``external_test''} target_type (string or list, optional): Type of target to use, ``Filename'', ``EF'', ``EDV'', ``ESV'', ``LargeIndex'', ``SmallIndex'', ``LargeFrame'', ``SmallFrame'', ``LargeTrace'', or ``SmallTrace'' Can also be a list to output a tuple with all specified target types. The targets represent: ``Filename'' (string): filename of video ``EF'' (float): ejection fraction ``EDV'' (float): end-diastolic volume ``ESV'' (float): end-systolic volume ``LargeIndex'' (int): index of large (diastolic) frame in video ``SmallIndex'' (int): index of small (systolic) frame in video ``LargeFrame'' (np.array shape=(3, height, width)): normalized large (diastolic) frame ``SmallFrame'' (np.array shape=(3, height, width)): normalized small (systolic) frame ``LargeTrace'' (np.array shape=(height, width)): left ventricle large (diastolic) segmentation value of 0 indicates pixel is outside left ventricle 1 indicates pixel is inside left ventricle ``SmallTrace'' (np.array shape=(height, width)): left ventricle small (systolic) segmentation value of 0 indicates pixel is outside left ventricle 1 indicates pixel is inside left ventricle Defaults to ``EF''. mean (int, float, or np.array shape=(3,), optional): means for all (if scalar) or each (if np.array) channel. Used for normalizing the video. Defaults to 0 (video is not shifted). std (int, float, or np.array shape=(3,), optional): standard deviation for all (if scalar) or each (if np.array) channel. Used for normalizing the video. Defaults to 0 (video is not scaled). length (int or None, optional): Number of frames to clip from video. If ``None'', longest possible clip is returned. Defaults to 16. period (int, optional): Sampling period for taking a clip from the video (i.e. every ``period''-th frame is taken) Defaults to 2. max_length (int or None, optional): Maximum number of frames to clip from video (main use is for shortening excessively long videos when ``length'' is set to None). If ``None'', shortening is not applied to any video. Defaults to 250. clips (int, optional): Number of clips to sample. Main use is for test-time augmentation with random clips. Defaults to 1. pad (int or None, optional): Number of pixels to pad all frames on each side (used as augmentation). and a window of the original size is taken. If ``None'', no padding occurs. Defaults to ``None''. noise (float or None, optional): Fraction of pixels to black out as simulated noise. If ``None'', no simulated noise is added. Defaults to ``None''. target_transform (callable, optional): A function/transform that takes in the target and transforms it. external_test_location (string): Path to videos to use for external testing. """ def __init__(self, root=None, split="train", target_type="EF", mean=0., std=1., length=16, period=2, max_length=250, clips=1, pad=None, noise=None, target_transform=None, external_test_location=None): if root is None: root = path super().__init__(root, target_transform=target_transform) self.split = split.upper() if not isinstance(target_type, list): target_type = [target_type] self.target_type = target_type self.mean = mean self.std = std self.length = length self.max_length = max_length self.period = period self.clips = clips self.pad = pad self.noise = noise self.target_transform = target_transform self.external_test_location = external_test_location self.fnames, self.outcome = [], [] if self.split == "EXTERNAL_TEST": self.fnames = sorted(os.listdir(self.external_test_location)) else: # Load video-level labels with open(f"{self.root}/FileList.csv") as f: data = pd.read_csv(f) data["Split"].map(lambda x: x.upper()) if self.split != "ALL": data = data[data["Split"] == self.split] self.header = data.columns.tolist() self.fnames = data["FileName"].tolist() self.fnames = [fn + ".avi" for fn in self.fnames if os.path.splitext(fn)[1] == ""] # Assume avi if no suffix self.outcome = data.values.tolist() # Check that files are present """ missing = set(self.fnames) - set(os.listdir(os.path.join(self.root, "Videos"))) if len(missing) != 0: print("{} videos could not be found in {}:".format(len(missing), os.path.join(self.root, "Videos"))) for f in sorted(missing): print("\t", f) raise FileNotFoundError(os.path.join(self.root, "Videos", sorted(missing)[0])) """ # Load traces self.frames = collections.defaultdict(list) self.trace = collections.defaultdict(_defaultdict_of_lists) with open(f"{self.root}/VolumeTracings.csv") as f: header = f.readline().strip().split(",") assert header == ["FileName", "X1", "Y1", "X2", "Y2", "Frame"] for line in f: filename, x1, y1, x2, y2, frame = line.strip().split(',') x1 = float(x1) y1 = float(y1) x2 = float(x2) y2 = float(y2) frame = int(frame) if frame not in self.trace[filename]: self.frames[filename].append(frame) self.trace[filename][frame].append((x1, y1, x2, y2)) for filename in self.frames: for frame in self.frames[filename]: self.trace[filename][frame] = np.array(self.trace[filename][frame]) # A small number of videos are missing traces; remove these videos keep = [len(self.frames[f]) >= 2 for f in self.fnames] self.fnames = [f for (f, k) in zip(self.fnames, keep) if k] self.outcome = [f for (f, k) in zip(self.outcome, keep) if k] def __getitem__(self, index): # Find filename of video if self.split == "EXTERNAL_TEST": video = os.path.join(self.external_test_location, self.fnames[index]) elif self.split == "CLINICAL_TEST": video = os.path.join(self.root, "ProcessedStrainStudyA4c", self.fnames[index]) else: video = os.path.join(self.root, "Videos", self.fnames[index]) # Load video into np.array video = echonet.utils.loadvideo(video).astype(np.float32) # Add simulated noise (black out random pixels) # 0 represents black at this point (video has not been normalized yet) if self.noise is not None: n = video.shape[1] * video.shape[2] * video.shape[3] ind = np.random.choice(n, round(self.noise * n), replace=False) f = ind % video.shape[1] ind //= video.shape[1] i = ind % video.shape[2] ind //= video.shape[2] j = ind video[:, f, i, j] = 0 # Apply normalization if isinstance(self.mean, (float, int)): video -= self.mean else: video -= self.mean.reshape(3, 1, 1, 1) if isinstance(self.std, (float, int)): video /= self.std else: video /= self.std.reshape(3, 1, 1, 1) # Set number of frames c, f, h, w = video.shape if self.length is None: # Take as many frames as possible length = f // self.period else: # Take specified number of frames length = self.length if self.max_length is not None: # Shorten videos to max_length length = min(length, self.max_length) if f < length * self.period: # Pad video with frames filled with zeros if too short # 0 represents the mean color (dark grey), since this is after normalization video = np.concatenate((video, np.zeros((c, length * self.period - f, h, w), video.dtype)), axis=1) c, f, h, w = video.shape # pylint: disable=E0633 if self.clips == "all": # Take all possible clips of desired length start = np.arange(f - (length - 1) * self.period) else: # Take random clips from video start = np.random.choice(f - (length - 1) * self.period, self.clips) # Gather targets target = [] for t in self.target_type: key = self.fnames[index] if t == "Filename": target.append(self.fnames[index]) elif t == "LargeIndex": # Traces are sorted by cross-sectional area # Largest (diastolic) frame is last target.append(int(self.frames[key][-1])) elif t == "SmallIndex": # Largest (diastolic) frame is first target.append(int(self.frames[key][0])) elif t == "LargeFrame": target.append(video[:, self.frames[key][-1], :, :]) elif t == "SmallFrame": target.append(video[:, self.frames[key][0], :, :]) elif t in ["LargeTrace", "SmallTrace"]: if t == "LargeTrace": t = self.trace[key][self.frames[key][-1]] else: t = self.trace[key][self.frames[key][0]] x1, y1, x2, y2 = t[:, 0], t[:, 1], t[:, 2], t[:, 3] x = np.concatenate((x1[1:], np.flip(x2[1:]))) y = np.concatenate((y1[1:], np.flip(y2[1:]))) r, c = skimage.draw.polygon(np.rint(y).astype(np.int), np.rint(x).astype(np.int), (video.shape[2], video.shape[3])) mask = np.zeros((video.shape[2], video.shape[3]), np.float32) mask[r, c] = 1 target.append(mask) else: if self.split == "CLINICAL_TEST" or self.split == "EXTERNAL_TEST": target.append(np.float32(0)) else: target.append(np.float32(self.outcome[index][self.header.index(t)])) if target != []: target = tuple(target) if len(target) > 1 else target[0] if self.target_transform is not None: target = self.target_transform(target) # Select clips from video video = tuple(video[:, s + self.period * np.arange(length), :, :] for s in start) if self.clips == 1: video = video[0] else: video = np.stack(video) if self.pad is not None: # Add padding of zeros (mean color of videos) # Crop of original size is taken out # (Used as augmentation) c, l, h, w = video.shape temp = np.zeros((c, l, h + 2 * self.pad, w + 2 * self.pad), dtype=video.dtype) temp[:, :, self.pad:-self.pad, self.pad:-self.pad] = video # pylint: disable=E1130 i, j = np.random.randint(0, 2 * self.pad, 2) video = temp[:, :, i:(i + h), j:(j + w)] return video, target def __len__(self): return len(self.fnames) def extra_repr(self) -> str: """Additional information to add at end of __repr__.""" lines = ["Target type: {target_type}", "Split: {split}"] return '\n'.join(lines).format(**self.__dict__) def _defaultdict_of_lists(): """Returns a defaultdict of lists. This is used to avoid issues with Windows (if this function is anonymous, the Echo dataset cannot be used in a dataloader). """ return collections.defaultdict(list) ## print("Done loading training data!") # define normalization layer to make sure output xi in an interval [ai, bi]: # define normalization layer to make sure output xi in an interval [ai, bi]: class IntervalNormalizationLayer(torch.nn.Module): def __init__(self): super().__init__() # new_output = [Tc, start_p, Emax, Emin, Rm, Ra, Vd] self.a = torch.tensor([0.4, 0., 0.5, 0.02, 0.005, 0.0001, 4.], dtype=torch.float32) #HR in 20-200->Tc in [0.3, 4] self.b = torch.tensor([1.7, 280., 3.5, 0.1, 0.1, 0.25, 16.], dtype=torch.float32) #taken out (initial conditions): a: 20, 5, 50; b: 400, 20, 100 def forward(self, inputs): sigmoid_output = torch.sigmoid(inputs) scaled_output = sigmoid_output * (self.b - self.a) + self.a return scaled_output class NEW3DCNN(nn.Module): def __init__(self, num_parameters): super(NEW3DCNN, self).__init__() self.conv1 = nn.Conv3d(3, 8, kernel_size=3, padding=1) self.batchnorm1 = nn.BatchNorm3d(8) self.conv2 = nn.Conv3d(8, 16, kernel_size=3, padding=1) self.batchnorm2 = nn.BatchNorm3d(16) self.conv3 = nn.Conv3d(16, 32, kernel_size=3, padding=1) self.batchnorm3 = nn.BatchNorm3d(32) self.conv4 = nn.Conv3d(32, 64, kernel_size=3, padding=1) self.batchnorm4 = nn.BatchNorm3d(64) self.conv5 = nn.Conv3d(64, 128, kernel_size=3, padding=1) self.batchnorm5 = nn.BatchNorm3d(128) self.pool = nn.AdaptiveAvgPool3d(1) self.fc1 = nn.Linear(128, 512) self.fc2 = nn.Linear(512, num_parameters) self.norm1 = IntervalNormalizationLayer() def forward(self, x): x = F.relu(self.batchnorm1(self.conv1(x))) x = F.max_pool3d(x, kernel_size=2, stride=2) x = F.relu(self.batchnorm2(self.conv2(x))) x = F.max_pool3d(x, kernel_size=2, stride=2) x = F.relu(self.batchnorm3(self.conv3(x))) x = F.max_pool3d(x, kernel_size=2, stride=2) x = F.relu(self.batchnorm4(self.conv4(x))) x = F.max_pool3d(x, kernel_size=2, stride=2) x = F.relu(self.batchnorm5(self.conv5(x))) x = self.pool(x) x = x.view(x.size(0), -1) x = F.relu(self.fc1(x)) x = self.fc2(x) x = self.norm1(x) return x # Define a neural network with one hidden layer class Interpolator(nn.Module): def __init__(self): super().__init__() self.fc1 = nn.Linear(6, 250).double() self.fc2 = nn.Linear(250, 2).double() def forward(self, x): x = torch.relu(self.fc1(x)) x = self.fc2(x) return x # Initialize the neural network net = Interpolator() net.load_state_dict(torch.load('final_model_weights/interp6_7param_weight.pt')) print("Done loading interpolator!") weights_path = 'final_model_weights/202_full_echonet_7param_Vloss_epoch_200_lr_0.001_weight_best_model.pt' model = NEW3DCNN(num_parameters = 7) model.load_state_dict(torch.load(weights_path)) model.to(device) ## PV loops #returns Plv at time t using Elastance(t) and Vlv(t)-Vd=x1 def Plv(volume, Emax, Emin, t, Tc, Vd): return Elastance(Emax,Emin,t, Tc)*(volume - Vd) #returns Elastance(t) def Elastance(Emax,Emin, t, Tc): t = t-int(t/Tc)*Tc #can remove this if only want 1st ED (and the 1st ES before) tn = t/(0.2+0.15*Tc) return (Emax-Emin)*1.55*(tn/0.7)**1.9/((tn/0.7)**1.9+1)*1/((tn/1.17)**21.9+1) + Emin def solve_ODE_for_volume(Rm, Ra, Emax, Emin, Vd, Tc, start_v, t): # the ODE from Simaan et al 2008 def heart_ode(y, t, Rs, Rm, Ra, Rc, Ca, Cs, Cr, Ls, Emax, Emin, Tc): x1, x2, x3, x4, x5 = y #here y is a vector of 5 values (not functions), at time t, used for getting (dy/dt)(t) P_lv = Plv(x1+Vd,Emax,Emin,t,Tc,Vd) dydt = [r(x2-P_lv)/Rm-r(P_lv-x4)/Ra, (x3-x2)/(Rs*Cr)-r(x2-P_lv)/(Cr*Rm), (x2-x3)/(Rs*Cs)+x5/Cs, -x5/Ca+r(P_lv-x4)/(Ca*Ra), (x4-x3-Rc*x5)/Ls] return dydt # RELU for diodes def r(u): return max(u, 0.) # Define fixed parameters Rs = 1.0 Rc = 0.0398 Ca = 0.08 Cs = 1.33 Cr = 4.400 Ls = 0.0005 startp = 75. # Initial conditions start_pla = float(start_v*Elastance(Emax, Emin, 0, Tc)) start_pao = startp start_pa = start_pao start_qt = 0 #aortic flow is Q_T and is 0 at ED, also see Fig5 in simaan2008dynamical y0 = [start_v, start_pla, start_pa, start_pao, start_qt] # Solve sol = odeint(heart_ode, y0, t, args = (Rs, Rm, Ra, Rc, Ca, Cs, Cr, Ls, Emax, Emin, Tc)) #t: list of values # volume is the first state variable plus theoretical zero pressure volume volumes = np.array(sol[:, 0]) + Vd return volumes def pvloop_simulator(Rm, Ra, Emax, Emin, Vd, Tc, start_v, animate=True): # Define initial parameters init_Emax = Emax # 3.0 # .5 to 3.5 init_Emin = Emin # 0.04 # .02 to .1 # init_Tc = Tc # .4 # .4 to 1.7 init_Vd = Vd # 10.0 # 0 to 25 # DUMMY VOLUME # def volume(t, Tc): # return 50*np.sin(2 * np.pi * t*(1/Tc))+100 # SOLVE the ODE model for the VOLUME CURVE N = 100 t = np.linspace(0, Tc*N, int(60000*N)) #np.linspace(1, 100, 1000000) volumes = solve_ODE_for_volume(Rm, Ra, Emax, Emin, Vd, Tc, start_v, t) # FUNCTIONS for PRESSURE CURVE vectorized_Elastance = np.vectorize(Elastance) vectorized_Plv = np.vectorize(Plv) def pressure(t, volume, Emax, Emin, Tc, Vd): return vectorized_Plv(volume, Emax, Emin, t, Tc, Vd) # calculate PRESSURE pressures = pressure(t, volumes, init_Emax, init_Emin, Tc, init_Vd) # Create the figure and the loop that we will manipulate fig, ax = plt.subplots() plt.ylim((0,220)) plt.xlim((0,250)) start = (N-2)*60000 end = (N)*60000 if animate: line = ax.plot(volumes[start:(start+1)], pressures[start:(start+1)], lw=1, color='b') point = ax.scatter(volumes[start:(start+1)], pressures[start:(start+1)], c="b", s=5) else: line = ax.plot(volumes[start:end], pressures[start:end], lw=1, color='b') plt.title('Predicted PI-SSL LV Pressure Volume Loop', fontsize=16) #plt.rcParams['fig.suptitle'] = -2.0 #ax.set_title(f'Mitral valve circuit resistance (Rm): {Rm} mmHg*s/ml \n Aortic valve circuit resistance (Ra): {Ra} mmHg*s/ml', fontsize=6) ax.set_xlabel('LV Volume (ml)') ax.set_ylabel('LV Pressure (mmHg)') # adjust the main plot to make room for the sliders # fig.subplots_adjust(left=0.25, bottom=0.25) def update(frame): # update to add more of the loop end = (N-2)*60000+1000 * frame x = volumes[start:end] y = pressures[start:end] ax.plot(x, y, lw=1, c='b') if animate: anim = animation.FuncAnimation(fig, partial(update), frames=100, interval=30) anim.save("prediction.mp4") return plt, Rm, Ra, Emax, Emin, Vd, Tc, start_v def pvloop_simulator_plot_only(Rm, Ra, Emax, Emin, Vd, Tc, start_v): plot,_,_,_,_,_,_,_ =pvloop_simulator(Rm, Ra, Emax, Emin, Vd, Tc, start_v, animate=False) plt.title('Simulated PI-SSL LV Pressure Volume Loop', fontsize=16) return plot ## Demo def generate_example(): # get random input data_path = 'EchoNet-Dynamic' image_data = Echo(root = data_path, split = 'all', target_type=['Filename','LargeIndex','SmallIndex']) image_loaded_data = DataLoader(image_data, batch_size=30, shuffle=True) val_data = next(iter(image_loaded_data)) #create_echo_clip(val_data,'test') val_seq = val_data[0] val_tensor = torch.tensor(val_seq, dtype=torch.float32) n=random.randint(0, 27) results = model(val_tensor)[n] filename = val_data[1][0][n] video = f"EchoNet-Dynamic/Videos/{filename}" plot, Rm, Ra, Emax, Emin, Vd,Tc, start_v = pvloop_simulator(Rm=round(results[4].item(),2), Ra=round(results[5].item(),2), Emax=results[2].item(), Emin=round(results[3].item(),2), Vd=round(results[6].item(),2), Tc=round(results[0].item(),2), start_v=round(results[1].item(),2)) video = video.replace("avi", "mp4") animated = "prediction.mp4" return video, animated, Rm, Ra, Emax, Emin, Vd, Tc, start_v title = "
Keying Kuang, Frances Dean, Jack B. Jedlicki, David Ouyang, Anthony Philippakis, David Sontag, Ahmed Alaa
We develop methodology for predicting digital twins from non-invasive cardiac ultrasound images in Non-Invasive Medical Digital Twins using Physics-Informed Self-Supervised Learning. Check out our code. \n \n We demonstrate the ability of our model to predict left ventricular pressure-volume loops using image data here. To run example predictions on samples from the EchoNet dataset, click the first button. \n \n Below you can input values of predicted parameters and output a simulated pressure-volume loop predicted by the Simaan et al 2008 hydraulic analogy model by pressing 'Run simulation.'
""" gr.Markdown(title) gr.Markdown(description) with gr.Blocks() as demo: # text gr.Markdown("