(self, g: dgl.DGLHeteroGraph)
| 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) |
nothing calls this directly
no test coverage detected