decagon-pytorch/src/triacontagon/model.py

from .data import Data, \
    EdgeType
import torch
from dataclasses import dataclass
from .weights import init_glorot
import types
from typing import List, \
    Dict, \
    Callable, \
    Tuple
from .util import _sparse_coo_tensor


@dataclass
class TrainingBatch(object):
    vertex_type_row: int
    vertex_type_column: int
    relation_type_index: int
    edges: torch.Tensor


class Model(torch.nn.Module):
    def __init__(self, data: Data, layer_dimensions: List[int],
        keep_prob: float,
        conv_activation: Callable[[torch.Tensor], torch.Tensor],
        dec_activation: Callable[[torch.Tensor], torch.Tensor],
        **kwargs) -> None:
        super().__init__(**kwargs)

        if not isinstance(data, Data):
            raise TypeError('data must be an instance of Data')

        if not isinstance(conv_activation, types.FunctionType):
            raise TypeError('conv_activation must be a function')

        if not isinstance(dec_activation, types.FunctionType):
            raise TypeError('dec_activation must be a function')

        self.data = data
        self.layer_dimensions = list(layer_dimensions)
        self.keep_prob = float(keep_prob)
        self.conv_activation = conv_activation
        self.dec_activation = dec_activation

        self.conv_weights = None
        self.dec_weights = None
        self.build()


    def build(self) -> None:
        self.conv_weights = torch.nn.ParameterDict()
        for i in range(len(self.layer_dimensions) - 1):
            in_dimension = self.layer_dimensions[i]
            out_dimension = self.layer_dimensions[i + 1]

            for _, et in self.data.edge_types.items():
                weight = init_glorot(in_dimension, out_dimension)
                self.conv_weights[et.vertex_type_row, et.vertex_type_column, i] = \
                    torch.nn.Parameter(weight)

        self.dec_weights = torch.nn.ParameterDict()
        for _, et in self.data.edge_types.items():
            global_interaction, local_variation = \
                et.decoder_factory(self.layer_dimensions[-1],
                    len(et.adjacency_matrices))
            self.dec_weights[et.vertex_type_row, et.vertex_type_column] = \
                torch.nn.ParameterList([
                    torch.nn.Parameter(global_interaction),
                    torch.nn.Parameter(local_variation)
                ])


    def convolve(self, batch: TrainingBatch) -> List[torch.Tensor]:
        edges = []
        cur_edges = batch.edges
        for _ in range(len(self.layer_dimensions) - 1):
            edges.append(cur_edges)
            key = (batch.vertex_type_row, batch.vertex_type_column)
            tot_conn = self.data.relation_types[key].total_connectivity
            cur_edges = _edges_for_rows(tot_conn, cur_edges[:, 1])


    def temporary_adjacency_matrix(self, adjacency_matrix: torch.Tensor,
        batch: TrainingBatch, total_connectivity: torch.Tensor) -> torch.Tensor:

        col = batch.vertex_type_column
        rows = batch.edges[:, 0]

        columns = batch.edges[:, 1].sum(dim=0).flatten()
        columns = torch.nonzero(columns)

        for i in range(len(self.layer_dimensions) - 1):
            pass # columns =
            # TODO: finish

        return None


    def temporary_adjacency_matrices(self, batch: TrainingBatch) -> Dict[Tuple[int, int], List[List[torch.Tensor]]]:

        col = batch.vertex_type_column
        batch.edges[:, 1]

        res = {}

        for _, et in self.data.edge_types.items():
            sum_nonzero = _nonzero_sum(et.adjacency_matrices)
            res[et.vertex_type_row, et.vertex_type_column] = \
                [ self.temporary_adjacency_matrix(adj_mat, batch,
                    et.total_connectivity) \
                        for adj_mat in et.adjacency_matrices ]

        return res

    def forward(self, initial_repr: List[torch.Tensor],
        batch: TrainingBatch) -> torch.Tensor:

        if not isinstance(initial_repr, list):
            raise TypeError('initial_repr must be a list')

        if len(initial_repr) != len(self.data.vertex_types):
            raise ValueError('initial_repr must contain representations for all vertex types')

        if not isinstance(batch, TrainingBatch):
            raise TypeError('batch must be an instance of TrainingBatch')

        adj_matrices = self.temporary_adjacency_matrices(batch)

        row_vertices = initial_repr[batch.vertex_type_row]
        column_vertices = initial_repr[batch.vertex_type_column]