code for creation of siamese samples.

create_siamese_samples.py

import itertools

import numpy as np

from diskdict import DiskDict
from diskarray import DiskArray
from basescript import BaseScript
from wordvecspace import WordVecSpaceMem

class SiameseSamples(BaseScript):
    def __init__(self):
        super(SiameseSamples, self).__init__()
        self.train_d = DiskArray(self.args.out_f, shape=(0,), dtype=self._get_dtype())
        self.wv = WordVecSpaceMem(self.args.wvspace)
        self.d = DiskDict(self.args.clusters_f)

    def _get_dtype(self):
        d_type = [
                    ('vec1', np.float32, 300),
                    ('vec2', np.float32, 300),
                    ('label', np.int, 1),
                ]

        return d_type

    def append_record(self, id1, id2, label):
        id1 = '<id:' + id1 + '>'
        id2 = '<id:' + id2 + '>'

        v1 = self.wv.get_word_vector(id1)
        v2 = self.wv.get_word_vector(id2)

        self.train_d.append((v1, v2, label))

    def run(self):
        for k, cluster in self.d.items():
            positives = cluster['positive']
            negatives = cluster['nagative']

            pairs = itertools.combinations(positives, 2)

            for pair in pairs:
                self.append_record(pair[0], pair[1], 0)

            self.create_negative_samples(positives, negatives)

        self.train_d.flush()

    def create_negative_samples(self, positives, negatives):
        for i in range(len(positives)):
            id1 = positives[i]

            for i in range(len(negatives)):
                id2 = negatives[i]
                self.append_record(id1, id2, 1)



    def define_args(self, parser):
        parser.add_argument('wvspace', help='vector space file')
        parser.add_argument('clusters_f', help='diskdict file which contains ids')
        parser.add_argument('out_f', help='output file to store labels')

if __name__ == '__main__':
    SiameseSamples().start()