MCPcopy Create free account
hub / github.com/ZJUDataIntelligence/Ultrasound-CLIP / forward

Method forward

graph_encoder.py:69–140  ·  view source on GitHub ↗
(self, g: dgl.DGLHeteroGraph)

Source from the content-addressed store, hash-verified

67 return pooled
68
69 def forward(self, g: dgl.DGLHeteroGraph):
70 device = next(self.parameters()).device
71 g = g.to(device)
72 try:
73 from torch.amp import autocast
74 from contextlib import nullcontext
75 autocast_ctx = autocast(device_type='cuda', enabled=False) if device.type == 'cuda' else nullcontext()
76 except Exception:
77 from contextlib import ExitStack as nullcontext
78 autocast_ctx = nullcontext()
79
80 with autocast_ctx:
81 self._ensure_type_embeddings(g, device)
82 self._ensure_convs(g, device)
83
84 try:
85 total_edges = 0
86 for et in g.canonical_etypes:
87 total_edges += g.num_edges(et)
88 except Exception:
89 total_edges = 0
90
91 batch_size = self._infer_batch_size(g)
92 if total_edges == 0:
93 return torch.zeros(batch_size, self.out_dim, device=device, dtype=torch.float32)
94
95 for ntype in g.ntypes:
96 num_nodes = g.num_nodes(ntype)
97 use_tid = 'tid' in g.nodes[ntype].data and num_nodes > 0
98 if use_tid and ntype in ('diagnosis', 'descriptor'):
99 tid = g.nodes[ntype].data['tid'].to(device)
100 if ntype == 'diagnosis':
101 feat = self.node_norm(self.diag_emb(tid).float())
102 else:
103 feat = self.node_norm(self.desc_emb(tid).float())
104 g.nodes[ntype].data['h'] = torch.nan_to_num(feat, nan=0.0, posinf=0.0, neginf=0.0)
105 else:
106 key = f"emb_{ntype}"
107 if key not in self.type_embeddings:
108 param = nn.Parameter(torch.zeros(self.hidden, device=device, dtype=torch.float32))
109 nn.init.xavier_uniform_(param.unsqueeze(0))
110 self.type_embeddings[key] = param
111 type_vec = self.type_embeddings[key]
112 h0 = type_vec.unsqueeze(0).expand(num_nodes, -1).contiguous()
113 g.nodes[ntype].data['h'] = h0.float()
114
115 for layer_idx in range(self.n_layers):
116 conv: HeteroGraphConv = self.convs[layer_idx]
117 h = {ntype: g.nodes[ntype].data['h'] for ntype in g.ntypes}
118 h = conv(g, h)
119 for ntype in h:
120 h_nt = torch.relu(h[ntype]).float()
121 h_nt = self.dropout(h_nt)
122 g.nodes[ntype].data['h'] = torch.nan_to_num(h_nt, nan=0.0, posinf=0.0, neginf=0.0)
123
124 pooled_list = []
125 for ntype in g.ntypes:
126 pooled = self._safe_mean_nodes(g, ntype)

Callers

nothing calls this directly

Calls 4

_ensure_convsMethod · 0.95
_infer_batch_sizeMethod · 0.95
_safe_mean_nodesMethod · 0.95

Tested by

no test coverage detected