text:
import time

import torch
from torch import nn
import tqdm
import numpy as np
from matplotlib import pyplot as plt

from ContentLoss.ResNetLoss import ResNetLoss
from ContentLoss.VGGLoss import VGGLoss
from Metric.PSNR import PSNR
import Metric.pytorch_msssim as torch_SSIM

def write_log_for_SRGAN(db_valid_PSNR_SSIM, file_name):
    f = open(file_name, 'w')
    f.write('write_log_for_SRGAN() \n\n')
    f.write(f"(1) valid PSNR per each epoch ->\n{str(db_valid_PSNR_SSIM['psnr'])} \n\n")
    f.write(f"(2) valid SSIM per each epoch ->\n{str(db_valid_PSNR_SSIM['ssim'])} \n")
    f.close()

def plot_log_for_SRGAN(db_valid_PSNR_SSIM, file_name):
    plt.clf()
    plt.title(file_name)
    plt.plot(range(1, len(db_valid_PSNR_SSIM['psnr']) + 1), np.array(db_valid_PSNR_SSIM['psnr']), label='valid_PSNR')
    plt.plot(range(1, len(db_valid_PSNR_SSIM['ssim']) + 1), np.array(db_valid_PSNR_SSIM['ssim']), label='valid_SSIM')
    plt.legend()
    plt.savefig(file_name)

def train_and_validate_SRGAN(generator, g_optimizer, discriminator, d_optimizer, epoch, device,
                                 train_loader, valid_loader, content_loss_function_name, tensorboard_writer):
    generator.to(device)
    discriminator.to(device)

    bce_loss_function = nn.BCELoss().to(device)

    content_loss_function = nn.MSELoss().to(device)
    if content_loss_function_name == 'vgg_loss_19_5_4':
        content_loss_function = VGGLoss(i=5, j=4, device=device, vgg_model_name='vgg19_bn')
    elif content_loss_function_name == 'res_loss_18_4':
        content_loss_function = ResNetLoss(i=4, device=device, resnet_model_name='resnet18')
    elif content_loss_function_name == 'res_loss_34_3':
        content_loss_function = ResNetLoss(i=3, device=device, resnet_model_name='resnet34')
    elif content_loss_function_name == 'res_loss_34_5':
        content_loss_function = ResNetLoss(i=5, device=device, resnet_model_name='resnet34')

    psnr_metric = PSNR()
    ssim_metric = torch_SSIM.MSSSIM()

    # (1) train phase
    train_d_loss, train_g_loss = 0, 0

    discriminator.train()
    generator.train()
    train_tqdm_loader = tqdm.tqdm(train_loader)
    for i, (lr_img, hr_img) in enumerate(train_tqdm_loader):
        lr_img = lr_img.to(device)
        hr_img = hr_img.to(device)

        # (1-1) train a discriminator model.
        d_optimizer.zero_grad()
        sr_img = generator(lr_img).detach()     # detach() <- for not back-propagation
        d_i_hr = discriminator(hr_img).flatten()
        d_i_sr = discriminator(sr_img).flatten()
        d_loss_real = bce_loss_function(d_i_hr, torch.ones_like(d_i_hr))    # Goal: D[I^{HR}]    ~= 1 (for real image)
        d_loss_fake = bce_loss_function(d_i_sr, torch.zeros_like(d_i_sr))   # Goal: D[G(I^{LR})] ~= 0 (for fake image)
        d_loss = d_loss_real + d_loss_fake      # maximize      E[log[D[I^{HR}]]] + E[log[(1 - D[G(I^{LR})])]]
                                                # for PyTorch implementation (PyTorch only has a minimizer!!!),
                                                # <=> minimize  E[- log[D[I^{HR}]]] + E[- log[1 - D[G(I^{LR})]]]
        train_d_loss += d_loss.item()
        d_loss.backward(retain_graph=True)
        d_optimizer.step()

        # (1-2) train a generator model.
        g_optimizer.zero_grad()
        hr_img = hr_img.detach()                # detach() <- for not back-propagation
        sr_img = generator(lr_img)
        d_i_sr = d_i_sr.detach()
        content_loss = content_loss_function(hr_img, sr_img)
        adversarial_loss = bce_loss_function(d_i_sr, torch.ones_like(d_i_sr))  # minimize  E[log[(1 - D[G(I^{LR})])]]
                                                                               # <=> minimize  - E[log[D[G(I^{LR})]]]
        g_loss = 1 * content_loss + 1e-3 * adversarial_loss

        train_g_loss += g_loss.item()
        g_loss.backward()
        g_optimizer.step()

        train_tqdm_loader.set_description(f'Train-SRGAN | Epoch: {epoch + 1} | '
                                          f'D_Loss: {train_d_loss / (i + 1): .4f} | '
                                          f'G_Loss: {train_g_loss / (i + 1): .4f} | ')

    time.sleep(0.01)

    # (2) validation phase
    generator.eval()
    valid_psnr = 0      # (2-1) PSNR
    valid_ssim = 0      # (2-2) SSIM
    valid_tqdm_loader = tqdm.tqdm(valid_loader)
    with torch.no_grad():
        for i, (lr_img, hr_img) in enumerate(valid_tqdm_loader):
            lr_img = lr_img.to(device)
            hr_img = hr_img.to(device)
            sr_img = generator(lr_img)
            # (2-1) PSNR
            psnr = psnr_metric(hr_img, sr_img)
            valid_psnr += psnr.item()

            # (2-2) SSIM
            ssim = ssim_metric(hr_img, sr_img)
            valid_ssim += ssim.item()

            valid_tqdm_loader.set_description(f'Valid-SRGAN | Epoch: {epoch + 1} | '
                                              f'PSNR: {psnr / (i + 1): .4f} | '
                                              f'PSNR: {psnr / (i + 1): .4f} | ')
        valid_psnr /= len(valid_tqdm_loader)
        valid_ssim /= len(valid_tqdm_loader)
        tensorboard_writer.add_scalar('PSNR/valid', valid_psnr, epoch + 1)
        tensorboard_writer.add_scalar('SSIM/valid', valid_ssim, epoch + 1)

    return valid_psnr, valid_ssim

max_stars_repo_name: unique-chan/SRGAN

max_issues_repo_name: unique-chan/SRGAN

max_forks_repo_name: unique-chan/SRGAN


text:
import cv2
import torch
import torch.nn.functional as F
import numpy as np
from argparse import ArgumentParser
from modules.tdmm_estimator import TDMMEstimator
from modules.flame_config import cfg as flame_cfg
from torch.utils.data import DataLoader
from frames_dataset import FramesDataset, ImageDataset
from tqdm import tqdm

if __name__ == "__main__":

    parser = ArgumentParser()
    parser.add_argument("--data_dir", default="", help="directory containing images to inference")
    parser.add_argument("--gpu", action="store_true", help="run the inference on gpu")
    parser.add_argument("--with_eye", action="store_true", help="use eye part for extracting texture")
    parser.add_argument("--tdmm_checkpoint", default=None, help="path to checkpoint of the tdmm estimator model to restore")

    opt = parser.parse_args()

    checkpoint = torch.load(opt.tdmm_checkpoint, map_location=torch.device('cpu'))

    tdmm = TDMMEstimator(flame_cfg)
    tdmm.load_state_dict(checkpoint['tdmm'], strict=False)
    if opt.gpu:
        tdmm = tdmm.cuda()

    dataset = ImageDataset(data_dir=opt.data_dir, meta_dir=None, augmentation_params=None)
    dataloader = DataLoader(dataset, batch_size=1, num_workers=1)
    tdmm.eval()

    for i, x in tqdm(enumerate(dataloader)):

        if opt.gpu:
            x['image'] = x['image'].cuda()

        codedict = tdmm.encode(x['image'])
        verts, transformed_verts, landmark_2d = tdmm.decode_flame(codedict)

        # extract albedo and rendering
        albedo = tdmm.extract_texture(x['image'], transformed_verts, with_eye=opt.with_eye)
        outputs = tdmm.render(transformed_verts, transformed_verts, albedo)

        image = x['image'].squeeze().permute(1, 2, 0)
        source = outputs['source'].squeeze().permute(1, 2, 0).detach()
        normal_map = outputs['source_normal_images'].squeeze().permute(1, 2, 0).detach()

        if opt.gpu:
            image = image.cpu()
            normal_map = normal_map.cpu()
            source = source.cpu()

        image = image.numpy()
        source = source.numpy()
        normal_map = normal_map.numpy()

        cv2.imshow('input', image[..., ::-1])
        cv2.imshow('source', source[..., ::-1])
        cv2.imshow('normal', normal_map)
        cv2.waitKey(0)

ext: py

max_stars_repo_name: shovelingpig/SAFA

max_issues_repo_name: shovelingpig/SAFA

max_forks_repo_name: shovelingpig/SAFA

reason: import numpy


text:
# pylint: disable=missing-module-docstring
# pylint: disable=missing-class-docstring
# pylint: disable=missing-function-docstring
# pylint: disable=no-self-use

import tempfile
import pytest
import numpy as np
from quantify_core.data.handling import set_datadir
from quantify_scheduler.schedules import timedomain_schedules as ts
from quantify_scheduler.compilation import determine_absolute_timing, qcompile
from quantify_scheduler.schemas.examples import utils
from .compiles_all_backends import _CompilesAllBackends

# FIXME to be replaced with fixture in tests/fixtures/schedule from !49 # pylint: disable=fixme
tmp_dir = tempfile.TemporaryDirectory()

# FIXME classmethods cannot use fixtures, these test are mixing testing style # pylint: disable=fixme
DEVICE_CONFIG = utils.load_json_example_scheme("transmon_test_config.json")

class TestRabiPulse(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "init_duration": 200e-6,
            "mw_G_amp": 0.5,
            "mw_D_amp": 0,
            "mw_frequency": 5.4e9,
            "mw_clock": "q0.01",
            "mw_port": "q0:mw",
            "mw_pulse_duration": 20e-9,
            "ro_pulse_amp": 0.1,
            "ro_pulse_duration": 1e-6,
            "ro_pulse_delay": 200e-9,
            "ro_pulse_port": "q0:res",
            "ro_pulse_clock": "q0.ro",
            "ro_pulse_frequency": 8e9,
            "ro_integration_time": 400e-9,
            "ro_acquisition_delay": 120e-9,
            "repetitions": 10,
        }

        cls.sched = ts.rabi_pulse_sched(**cls.sched_kwargs)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        sched = determine_absolute_timing(self.sched)
        # test that the right operations are added and timing is as expected.
        labels = ["qubit reset", "Rabi_pulse", "readout_pulse", "acquisition"]
        t2 = (
            self.sched_kwargs["init_duration"]
            + self.sched_kwargs["mw_pulse_duration"]
            + self.sched_kwargs["ro_pulse_delay"]
        )
        t3 = t2 + self.sched_kwargs["ro_acquisition_delay"]
        abs_times = [0, self.sched_kwargs["init_duration"], t2, t3]

        for i, constr in enumerate(sched.timing_constraints):
            assert constr["label"] == labels[i]
            assert constr["abs_time"] == abs_times[i]

    def test_compiles_device_cfg_only(self, load_example_transmon_config):
        # assert that files properly compile
        qcompile(self.sched, load_example_transmon_config())

class TestRabiSched(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "pulse_amp": 0.2,
            "pulse_duration": 20e-9,
            "frequency": 5.442e9,
            "qubit": "q0",
            "port": None,
            "clock": None,
            "repetitions": 10,
        }

        cls.sched = ts.rabi_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        labels = ["Reset 0", "Rabi_pulse 0", "Measurement 0"]
        abs_times = [0, 200e-6, 200e-6 + 20e-9]

        assert len(self.sched.timing_constraints) == len(labels)
        for i, constr in enumerate(self.sched.timing_constraints):
            assert constr["label"] == labels[i]
            assert constr["abs_time"] == abs_times[i]

    def test_rabi_pulse_ops(self):
        rabi_op_hash = self.sched.timing_constraints[1]["operation_repr"]
        rabi_pulse = self.sched.operations[rabi_op_hash]["pulse_info"][0]
        assert rabi_pulse["G_amp"] == 0.2
        assert rabi_pulse["D_amp"] == 0
        assert rabi_pulse["duration"] == 20e-9
        assert self.sched.resources["q0.01"]["freq"] == 5.442e9

    def test_batched_variant_single_val(self, load_example_transmon_config):
        sched = ts.rabi_sched(
            pulse_amp=[0.5],
            pulse_duration=20e-9,
            frequency=5.442e9,
            qubit="q0",
            port=None,
            clock=None,
        )
        sched = qcompile(sched, load_example_transmon_config())

        # test that the right operations are added and timing is as expected.
        labels = ["Reset 0", "Rabi_pulse 0", "Measurement 0"]
        assert len(sched.timing_constraints) == len(labels)
        for i, constr in enumerate(sched.timing_constraints):
            assert constr["label"] == labels[i]

        rabi_op_hash = sched.timing_constraints[1]["operation_repr"]
        rabi_pulse = sched.operations[rabi_op_hash]["pulse_info"][0]
        assert rabi_pulse["G_amp"] == 0.5
        assert rabi_pulse["D_amp"] == 0
        assert rabi_pulse["duration"] == 20e-9

    def test_batched_variant_amps(self, load_example_transmon_config):

        amps = np.linspace(-0.5, 0.5, 5)
        sched = ts.rabi_sched(
            pulse_amp=amps,
            pulse_duration=20e-9,
            frequency=5.442e9,
            qubit="q0",
            port=None,
            clock=None,
        )
        sched = qcompile(sched, load_example_transmon_config())

        # test that the right operations are added and timing is as expected.
        labels = []
        for j in range(5):
            labels += [f"Reset {j}", f"Rabi_pulse {j}", f"Measurement {j}"]
        assert len(sched.timing_constraints) == len(labels)
        for i, constr in enumerate(sched.timing_constraints):
            assert constr["label"] == labels[i]

        for i, exp_amp in enumerate(amps):
            rabi_op_hash = sched.timing_constraints[3 * i + 1]["operation_repr"]
            rabi_pulse = sched.operations[rabi_op_hash]["pulse_info"][0]
            assert rabi_pulse["G_amp"] == exp_amp
            assert rabi_pulse["D_amp"] == 0
            assert rabi_pulse["duration"] == 20e-9

    def test_batched_variant_durations(self, load_example_transmon_config):

        durations = np.linspace(3e-9, 30e-9, 6)
        sched = ts.rabi_sched(
            pulse_amp=0.5,
            pulse_duration=durations,
            frequency=5.442e9,
            qubit="q0",
            port=None,
            clock=None,
        )
        sched = qcompile(sched, load_example_transmon_config())

        # test that the right operations are added and timing is as expected.
        labels = []
        for j in range(6):
            labels += [f"Reset {j}", f"Rabi_pulse {j}", f"Measurement {j}"]

        assert len(sched.timing_constraints) == len(labels)
        for i, constr in enumerate(sched.timing_constraints):
            assert constr["label"] == labels[i]

    def test_batched_variant_incompatible(self):
        with pytest.raises(ValueError):
            _ = ts.rabi_sched(
                pulse_amp=np.linspace(-0.3, 0.5, 3),
                pulse_duration=np.linspace(5e-9, 19e-9, 8),
                frequency=5.442e9,
                qubit="q0",
                port=None,
                clock=None,
            )

    def test_correct_inference_of_port_clock(self):
        # operation 1 is tested in test_timing to be the Rabi pulse
        op_name = self.sched.timing_constraints[1]["operation_repr"]
        rabi_op = self.sched.operations[op_name]
        assert rabi_op["pulse_info"][0]["port"] == "q0:mw"
        assert rabi_op["pulse_info"][0]["clock"] == "q0.01"

class TestT1Sched(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "times": np.linspace(0, 80e-6, 21),
            "qubit": "q0",
            "repetitions": 10,
        }

        cls.sched = ts.t1_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        labels = []
        label_tmpl = ["Reset {}", "pi {}", "Measurement {}"]
        for i in range(len(self.sched_kwargs["times"])):
            labels += [l.format(i) for l in label_tmpl]

        for i, constr in enumerate(self.sched.timing_constraints):
            assert constr["label"] == labels[i]
            if (i - 2) % 3 == 0:  # every measurement operation
                assert constr["rel_time"] == self.sched_kwargs["times"][i // 3]

    # pylint: disable=no-self-use
    def test_sched_float_times(self, load_example_transmon_config):
        sched_kwargs = {
            "times": 3e-6,  # a floating point time
            "qubit": "q0",
        }

        sched = ts.t1_sched(**sched_kwargs)
        sched = qcompile(sched, load_example_transmon_config())

    def test_operations(self):
        assert len(self.sched.operations) == 2 + 21  # init, pi and 21*measure

class TestRamseySchedDetuning(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "times": np.linspace(4.0e-6, 80e-6, 20),
            "qubit": "q0",
            "artificial_detuning": 250e3,
            "repetitions": 10,
        }

        cls.sched = ts.ramsey_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        for i, constr in enumerate(self.sched.timing_constraints):
            if i % 4 == 0:
                assert constr["label"][:5] == "Reset"
            if (i - 2) % 4 == 0:  # every second pi/2 operation
                assert constr["rel_time"] == self.sched_kwargs["times"][i // 4]
            if (i - 3) % 4 == 0:
                assert constr["label"][:11] == "Measurement"

    # pylint: disable=no-self-use
    def test_sched_float_times(self, load_example_transmon_config):
        sched_kwargs = {
            "times": 3e-6,  # a floating point time
            "qubit": "q0",
            "artificial_detuning": 250e3,
        }

        sched = ts.ramsey_sched(**sched_kwargs)
        sched = qcompile(sched, load_example_transmon_config())
        assert any(op["rel_time"] == 3e-6 for op in sched.timing_constraints)

    def test_operations(self):
        assert len(self.sched.operations) == 2 + len(self.sched_kwargs["times"]) * 2

class TestRamseySched(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "times": np.linspace(4.0e-6, 80e-6, 20),
            "qubit": "q0",
            "repetitions": 10,
        }

        cls.sched = ts.ramsey_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        for i, constr in enumerate(self.sched.timing_constraints):
            if i % 4 == 0:
                assert constr["label"][:5] == "Reset"
            if (i - 2) % 4 == 0:  # every second pi/2 operation
                assert constr["rel_time"] == self.sched_kwargs["times"][i // 4]
            if (i - 3) % 4 == 0:
                assert constr["label"][:11] == "Measurement"

    # pylint: disable=no-self-use
    def test_sched_float_times(self, load_example_transmon_config):
        sched_kwargs = {
            "times": 3e-6,  # a floating point time
            "qubit": "q0",
        }

        sched = ts.ramsey_sched(**sched_kwargs)
        sched = qcompile(sched, load_example_transmon_config())

    def test_operations(self):
        assert (
            len(self.sched.operations) == 3 + 20
        )  # init, x90, Rxy(90,0) and 20 * measure

class TestEchoSched(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "times": np.linspace(4.0e-6, 80e-6, 20),
            "qubit": "q0",
            "repetitions": 10,
        }

        cls.sched = ts.echo_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    # pylint: disable=no-self-use
    def test_sched_float_times(self, load_example_transmon_config):
        sched_kwargs = {
            "times": 3e-6,  # a floating point time
            "qubit": "q0",
        }

        sched = ts.echo_sched(**sched_kwargs)
        sched = qcompile(sched, load_example_transmon_config())

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        for i, constr in enumerate(self.sched.timing_constraints):
            if i % 5 == 0:
                assert constr["label"][:5] == "Reset"
            if (i - 2) % 5 == 0:  # every second pi/2 operation
                assert constr["rel_time"] == self.sched_kwargs["times"][i // 5] / 2
            if (i - 3) % 5 == 0:  # every second pi/2 operation
                assert constr["rel_time"] == self.sched_kwargs["times"][i // 5] / 2
            if (i - 4) % 5 == 0:
                assert constr["label"][:11] == "Measurement"

    def test_operations(self):
        # 4 for an echo
        assert len(self.sched.operations) == 23  # init, x90, X and 20x measure

class TestAllXYSched(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {"qubit": "q0", "repetitions": 10}

        cls.sched = ts.allxy_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_repetitions(self):
        assert self.sched.repetitions == self.sched_kwargs["repetitions"]

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        for i, constr in enumerate(self.sched.timing_constraints):
            if i % 4 == 0:
                assert constr["label"][:5] == "Reset"
            if (i - 3) % 4 == 0:
                assert constr["label"][:11] == "Measurement"

    def test_operations(self):
        # 6 +21 operations (x90, y90, X180, Y180, idle, reset, 21*measurement)
        assert len(self.sched.operations) == 6 + 21

class TestAllXYSchedElement(_CompilesAllBackends):
    @classmethod
    def setup_class(cls):
        set_datadir(tmp_dir.name)
        cls.sched_kwargs = {
            "qubit": "q0",
            "element_select_idx": 4,
        }

        cls.sched = ts.allxy_sched(**cls.sched_kwargs)
        cls.sched = qcompile(cls.sched, DEVICE_CONFIG)

    def test_timing(self):
        # test that the right operations are added and timing is as expected.
        for i, constr in enumerate(self.sched.timing_constraints):
            if i % 4 == 0:
                assert constr["label"][:5] == "Reset"
            if (i - 3) % 4 == 0:
                assert constr["label"][:11] == "Measurement"

    def test_operations(self):
        # 4 operations (X180, Y180, reset, measurement)
        assert len(self.sched.operations) == 4

max_stars_repo_name: quantify-os/quantify-scheduler

max_issues_repo_name: quantify-os/quantify-scheduler

max_forks_repo_name: quantify-os/quantify-scheduler


text:
import os
import random
import spacy
import string
from textwrap import wrap
from typing import List, Dict

import ffmpeg
from pydub import AudioSegment
from tqdm import tqdm
import numpy as np

from animation import animation_cache, SceneAnimation

from script_constants import Location, Character, Action, location_map, character_map, character_location_map, \
  audio_emotions, character_emotions, objection_emotions, shake_emotions, hold_it_emotions

from emotions import EmotionModel

from comments import Comment, EmotionComment, Author

class PhoenixEngine(object):
  def __init__(
    self,
    theme='classic',
    emotion_model='mrm8488/t5-base-finetuned-emotion',
    sentence_model='en_core_web_sm',
    emotion_threshold=0.5,
    music_min_scene_duration=4,
    fps=18,
    video_codec='libx264',
    audio_codec='aac',
    video_crf=23,
    lag_frames=25,
    default_animation_length=11,
    scaling_factor=2.0,
    assets_folder='./assets',
    cache_folder='./cache',
  ):
    self.theme = theme
    self.emotion_model = emotion_model
    self.sentence_model = sentence_model
    self.emotion_threshold = emotion_threshold
    self.music_min_scene_duration = music_min_scene_duration
    self.fps = fps
    self.video_codec = video_codec
    self.audio_codec = audio_codec
    self.video_crf = video_crf
    self.lag_frames = lag_frames
    self.default_animation_length = default_animation_length
    self.scaling_factor = scaling_factor
    self.assets_folder = os.path.join(assets_folder, self.theme)
    self.cache_folder = cache_folder
    # mrm8488/t5-base-finetuned-emotion
    self.emo = EmotionModel(
      self.emotion_model
    )

    # en_core_web_sm
    self.nlp = spacy.load(
      sentence_model
    )

  def animate(
    self,
    comments: List[Comment],
    output_filename
  ):
    os.makedirs(os.path.dirname(output_filename), exist_ok=True)

    emotion_comments = self.emo.detect_emotions(comments)

    scene_config = self._configure_scene(emotion_comments)

    if not os.path.exists(self.cache_folder):
      os.mkdir(self.cache_folder)

    sound_effects, video_path = self._render_video(
      scene_config
    )

    audio_path = self._render_audio(
      sound_effects
    )

    video_input = ffmpeg.input(video_path)
    audio_input = ffmpeg.input(audio_path)

    process = (
      ffmpeg.output(
        video_input,
        audio_input,
        output_filename,
        pix_fmt='yuv420p',
        vcodec=self.video_codec,
        r=self.fps,
        acodec=self.audio_codec,
        crf=self.video_crf
      )
        .overwrite_output()
        .run_async(pipe_stdin=True)
    )
    process.wait()

  def _configure_scene(self, comments: List[EmotionComment]):
    # 30 chars per line, 3 lines, but we must subtract 3 for the final potential "..."
    wrap_threshold = (3 * 30) - 3
    scene = []

    for comment in comments:
      tokens = self.nlp(comment.body)
      sentences = [sent.string.strip() for sent in tokens.sents]
      joined_sentences, current_sentence = [], None

      for sentence in sentences:
        if len(sentence) > wrap_threshold:
          text_chunks = []
          text_wrap = wrap(sentence, wrap_threshold)

          for idx, chunk in enumerate(text_wrap):
            if idx != len(text_wrap) - 1 and chunk[-1] not in string.punctuation:
              chunk_text = f"{chunk}..."
            else:
              chunk_text = chunk
            text_chunks.append(chunk_text)
          joined_sentences.extend(text_chunks)
          current_sentence = None
        else:
          if current_sentence is not None and len(current_sentence) + len(sentence) + 1 <= wrap_threshold:
            current_sentence += " " + sentence
          else:
            if current_sentence is not None:
              joined_sentences.append(current_sentence)
            current_sentence = sentence

      if current_sentence is not None:
        joined_sentences.append(current_sentence)

      character_block = []
      character = comment.author.character
      emotion = comment.emotion
      emotion_score = comment.score

      if emotion is None or emotion_score <= self.emotion_threshold:
        emotion = 'normal'

      character_emotion = random.choice(character_emotions[character][emotion])

      for idx, chunk in enumerate(joined_sentences):
        character_block.append(
          {
            "character": character,
            "name": comment.author.name,
            "text": chunk,
            "emotion": character_emotion,
            "emotion_class": emotion,
            "emotion_class_score": emotion_score
          }
        )
      scene.append(character_block)
    formatted_scenes = []
    last_audio = audio_emotions['normal']
    change_audio = True
    last_emotion = None
    audio_duration = 0

    for character_block in scene:
      scene_objs = []
      character = character_block[0]["character"]
      emotion = character_block[0]["emotion_class"]

      if last_emotion is None:
        last_emotion = emotion
      if emotion in objection_emotions:
        scene_objs.append(
          {
            "character": character,
            "action": Action.OBJECTION,
          }
        )

        if last_audio != audio_emotions['objection']:
          last_audio = audio_emotions['objection']
          change_audio = True
      elif emotion in shake_emotions:
        scene_objs.append(
          {
            "character": character,
            "action": Action.SHAKE_EFFECT,
          }
        )
      elif emotion in hold_it_emotions:
        scene_objs.append(
          {
            "character": character,
            "action": Action.HOLD_IT,
          }
        )

        if last_audio != audio_emotions['holdit']:
          last_audio = audio_emotions['holdit']
          change_audio = True
      elif emotion != last_emotion \
        and audio_duration >= self.music_min_scene_duration:
        if last_audio != audio_emotions[emotion]:
          last_audio = audio_emotions[emotion]
          change_audio = True
          last_emotion = emotion

      for obj in character_block:
        scene_objs.append(
          {
            "character": obj["character"],
            "action": Action.TEXT,
            "emotion": obj["emotion"],
            "text": obj["text"],
            "name": obj["name"],
          }
        )
      formatted_scene = {
        "location": character_location_map[character],
        "scene": scene_objs,
      }

      if change_audio:
        formatted_scene["audio"] = last_audio
        change_audio = False
        audio_duration = 0

      audio_duration += 1
      formatted_scenes.append(formatted_scene)
    return formatted_scenes

  def _render_video(self, scene_configs):
    video_path = os.path.join(self.cache_folder, 'video.mp4')
    sound_effects = []
    process = None
    for scene_config in tqdm(scene_configs, desc='creating video', total=len(scene_configs)):
      scene_animations, scene_sfx = self._process_scene(scene_config)
      for animation in scene_animations:
        for frame in animation.frames:
          frame_array = np.array(frame)[:, :, :3]
          if process is None:
            height, width, channels = frame_array.shape
            video_input = ffmpeg.input(
              'pipe:',
              format='rawvideo',
              pix_fmt='rgb24',
              s=f'{width}x{height}',
              r=self.fps
            )
            process = (
              ffmpeg.output(
                video_input,
                video_path,
                pix_fmt='yuv420p',
                vcodec=self.video_codec,
                r=self.fps,
                crf=self.video_crf
              )
                .overwrite_output()
                .run_async(pipe_stdin=True)
            )
          process.stdin.write(
            frame_array.astype(np.uint8).tobytes()
          )
      sound_effects.extend(scene_sfx)
    process.stdin.close()
    process.wait()

    animation_cache.clear()

    return sound_effects, video_path

  def _render_audio(self, sound_effects: List[Dict]):
    audio_se = AudioSegment.empty()
    bip = AudioSegment.from_wav(
      f"{self.assets_folder}/sfx general/sfx-blipmale.wav"
    ) + AudioSegment.silent(duration=50)
    blink = AudioSegment.from_wav(f"{self.assets_folder}/sfx general/sfx-blink.wav")
    blink -= 10
    badum = AudioSegment.from_wav(f"{self.assets_folder}/sfx general/sfx-fwashing.wav")
    long_bip = bip * 100
    long_bip -= 10
    spf = 1 / self.fps * 1000
    pheonix_objection = AudioSegment.from_mp3(f"{self.assets_folder}/Phoenix - objection.mp3")
    edgeworth_objection = AudioSegment.from_mp3(
      f"{self.assets_folder}/Edgeworth - (English) objection.mp3"
    )
    default_objection = AudioSegment.from_mp3(f"{self.assets_folder}/Payne - Objection.mp3")

    # loop through all sfx which are not music tracks and combine them into a single track
    for obj in tqdm(sound_effects, total=len(sound_effects), desc='creating sound effects'):
      obj_type = obj["_type"]
      obj_length = 0

      if 'length' in obj:
        obj_length = obj["length"]

      obj_duration = int(obj_length * spf)

      if obj_type == "silence":
        audio_se += AudioSegment.silent(duration=obj_duration)
      elif obj_type == "bip":
        obj_duration -= len(blink)
        audio_se += blink + long_bip[:obj_duration]
      elif obj_type == "objection":
        obj_character = obj["character"]

        if obj_character == "phoenix":
          audio_se += pheonix_objection[:obj_duration]
        elif obj_character == "edgeworth":
          audio_se += edgeworth_objection[:obj_duration]
        else:
          audio_se += default_objection[:obj_duration]
      elif obj_type == "shock":
        audio_se += badum[:obj_duration]

    music_tracks = []
    len_counter = 0
    # loop through all music tracks and determine their length based on sound effects between them
    for obj in sound_effects:
      if obj["_type"] == "bg":
        if len(music_tracks) > 0:
          music_tracks[-1]["length"] = len_counter
          len_counter = 0
        music_tracks.append({"src": obj["src"]})
      else:
        len_counter += obj["length"]

    if len(music_tracks) > 0 and len_counter > 0:
      music_tracks[-1]["length"] = len_counter

    music_se = AudioSegment.empty()
    # create music track based on computed lengths and combine tracks together
    for track in tqdm(music_tracks, total=len(music_tracks), desc='creating music'):
      track_length = track["length"]
      track_duration = int(track_length * spf)
      music_se += AudioSegment.from_mp3(track["src"])[:track_duration]

    final_se = audio_se.overlay(music_se)
    audio_path = os.path.join(self.cache_folder, 'audio.mp3')
    final_se.export(audio_path, format="mp3")
    return audio_path

  def _process_scene(self, scene):
    sound_effects = []
    scene_animations = []
    bg = animation_cache.get_anim_img(
      f'{self.assets_folder}/{location_map[scene["location"]]}',
      scaling_factor=self.scaling_factor
    )
    arrow = animation_cache.get_anim_img(
      f"{self.assets_folder}/arrow.png",
      x=235,
      y=170,
      w=15,
      h=15,
      key_x=5,
      scaling_factor=self.scaling_factor
    )
    textbox = animation_cache.get_anim_img(
      f"{self.assets_folder}/textbox4.png",
      w=bg.w // self.scaling_factor,
      scaling_factor=self.scaling_factor
    )
    name_text_font_size = 10
    name_text_x = 4
    name_text_y = 115
    text_box_font_size = 15
    text_box_x = 5
    text_box_y = 130
    text_box_max_line_count = 32
    bench = None

    if scene["location"] == Location.COURTROOM_LEFT:
      bench = animation_cache.get_anim_img(
        f"{self.assets_folder}/logo-left.png",
        scaling_factor=self.scaling_factor
      )
    elif scene["location"] == Location.COURTROOM_RIGHT:
      bench = animation_cache.get_anim_img(
        f"{self.assets_folder}/logo-right.png",
        scaling_factor=self.scaling_factor
      )
    elif scene["location"] == Location.WITNESS_STAND:
      bench = animation_cache.get_anim_img(
        f"{self.assets_folder}/witness_stand.png",
        w=bg.w // self.scaling_factor,
        scaling_factor=self.scaling_factor
      )
      bench.y = bg.h - bench.h

    if "audio" in scene:
      sound_effects.append({"_type": "bg", "src": f'{self.assets_folder}/{scene["audio"]}.mp3'})

    current_frame = 0
    current_character_name = None
    text = None

    for obj in scene["scene"]:
      if "character" in obj:
        _dir = f'{self.assets_folder}/{character_map[obj["character"]]}'
        current_character_name = str(obj["character"])
        character_name = animation_cache.get_anim_text(
          current_character_name,
          font_path=f"{self.assets_folder}/igiari/Igiari.ttf",
          font_size=name_text_font_size,
          x=name_text_x,
          y=name_text_y,
          scaling_factor=self.scaling_factor
        )
        default = "normal" if "emotion" not in obj else obj["emotion"]
        default_path = (
          f"{_dir}/{current_character_name.lower()}-{default}(a).gif"
        )

        if not os.path.isfile(default_path):
          default_path = (
            f"{_dir}/{current_character_name.lower()}-{default}.gif"
          )

        default_character = animation_cache.get_anim_img(
          default_path,
          half_speed=True,
          scaling_factor=self.scaling_factor
        )

        if "(a)" in default_path:
          talking_character = animation_cache.get_anim_img(
            default_path.replace("(a)", "(b)"),
            half_speed=True,
            scaling_factor=self.scaling_factor
          )
        else:
          talking_character = animation_cache.get_anim_img(
            default_path,
            half_speed=True,
            scaling_factor=self.scaling_factor
          )

      if "emotion" in obj:
        default = obj["emotion"]
        default_path = (
          f"{_dir}/{current_character_name.lower()}-{default}(a).gif"
        )

        if not os.path.isfile(default_path):
          default_path = (
            f"{_dir}/{current_character_name.lower()}-{default}.gif"
          )

        default_character = animation_cache.get_anim_img(
          default_path,
          half_speed=True,
          scaling_factor=self.scaling_factor
        )

        if "(a)" in default_path:
          talking_character = animation_cache.get_anim_img(
            default_path.replace("(a)", "(b)"),
            half_speed=True,
            scaling_factor=self.scaling_factor
          )
        else:
          talking_character = animation_cache.get_anim_img(
            default_path,
            half_speed=True,
            scaling_factor=self.scaling_factor
          )

      if "action" in obj and (
        obj["action"] == Action.TEXT
        or obj["action"] == Action.TEXT_SHAKE_EFFECT
      ):
        character = talking_character
        _text = split_str_into_newlines(obj["text"], text_box_max_line_count)
        _colour = None if "colour" not in obj else obj["colour"]
        text = animation_cache.get_anim_text(
          _text,
          font_path=f"{self.assets_folder}/igiari/Igiari.ttf",
          font_size=text_box_font_size,
          x=text_box_x,
          y=text_box_y,
          typewriter_effect=True,
          colour=_colour,
          scaling_factor=self.scaling_factor
        )
        num_frames = len(_text) + self.lag_frames
        _character_name = character_name

        if "name" in obj:
          _character_name = animation_cache.get_anim_text(
            obj["name"],
            font_path=f"{self.assets_folder}/igiari/Igiari.ttf",
            font_size=name_text_font_size,
            x=name_text_x,
            y=name_text_y,
            scaling_factor=self.scaling_factor
          )

        if obj["action"] == Action.TEXT_SHAKE_EFFECT:
          bg.shake_effect = True
          character.shake_effect = True

          if bench is not None:
            bench.shake_effect = True

          textbox.shake_effect = True

        scene_animations.append(
          SceneAnimation(
            [bg, character, bench, textbox, _character_name, text],
            length=len(_text) - 1,
            start_frame=current_frame
          )
        )
        sound_effects.append({"_type": "bip", "length": len(_text) - 1})

        if obj["action"] == Action.TEXT_SHAKE_EFFECT:
          bg.shake_effect = False
          character.shake_effect = False

          if bench is not None:
            bench.shake_effect = False

          textbox.shake_effect = False
        text.typewriter_effect = False
        character = default_character
        scene_animations.append(
          SceneAnimation(
            [bg, character, bench, textbox, _character_name, text, arrow],
            length=self.lag_frames,
            start_frame=len(_text) - 1
          )
        )
        current_frame += num_frames
        sound_effects.append({"_type": "silence", "length": self.lag_frames})

      elif "action" in obj and obj["action"] == Action.SHAKE_EFFECT:
        character = default_character
        bg.shake_effect = True
        character.shake_effect = True

        if bench is not None:
          bench.shake_effect = True

        textbox.shake_effect = True

        if text is not None:
          scene_objs = [
            bg,
            character,
            bench,
            textbox,
            character_name,
            text,
            arrow,
          ]
        else:
          scene_objs = [bg, character, bench]

        scene_animations.append(
          SceneAnimation(
            scene_objs,
            length=self.lag_frames,
            start_frame=current_frame
          )
        )
        sound_effects.append({"_type": "shock", "length": self.lag_frames})
        current_frame += self.lag_frames
        bg.shake_effect = False
        character.shake_effect = False

        if bench is not None:
          bench.shake_effect = False

        textbox.shake_effect = False

      elif "action" in obj and obj["action"] == Action.OBJECTION:
        effect_image = animation_cache.get_anim_img(
          f"{self.assets_folder}/objection.gif",
          shake_effect=True,
          scaling_factor=self.scaling_factor
        )
        character = default_character
        scene_animations.append(
          SceneAnimation(
            [bg, character, bench, effect_image],
            length=self.default_animation_length,
            start_frame=current_frame
          )
        )
        scene_animations.append(
          SceneAnimation(
            [bg, character, bench],
            length=self.default_animation_length,
            start_frame=current_frame
          )
        )

        sound_effects.append(
          {
            "_type": "objection",
            "character": current_character_name.lower(),
            "length": 2 * self.default_animation_length,
          }
        )
        current_frame += self.default_animation_length

      elif "action" in obj and obj["action"] == Action.HOLD_IT:
        effect_image = animation_cache.get_anim_img(
          f"{self.assets_folder}/holdit.gif",
          shake_effect=True,
          scaling_factor=self.scaling_factor
        )
        character = default_character
        scene_animations.append(
          SceneAnimation(
            [bg, character, bench, effect_image],
            length=self.default_animation_length,
            start_frame=current_frame
          )
        )
        scene_animations.append(
          SceneAnimation(
            [bg, character, bench],
            length=self.default_animation_length,
            start_frame=current_frame
          )
        )
        sound_effects.append(
          {
            "_type": "objection",
            "character": current_character_name.lower(),
            "length": 2 * self.default_animation_length,
          }
        )
        current_frame += self.default_animation_length

      else:
        character = default_character
        _length = self.lag_frames

        if "length" in obj:
          _length = obj["length"]
        if "repeat" in obj:
          character.repeat = obj["repeat"]

        scene_animations.append(
          SceneAnimation(
            [bg, character, bench],
            length=_length,
            start_frame=current_frame
          )
        )
        character.repeat = True
        sound_effects.append({"_type": "silence", "length": _length})
        current_frame += _length

    return scene_animations, sound_effects

  def get_characters(self, most_common: List):
    # this may be based on theme in the future, don't make static
    characters = {Character.PHOENIX: most_common[0]}

    if len(most_common) > 0:
      characters[Character.EDGEWORTH] = most_common[1]

      if len(most_common) > 1:
        for character in most_common[2:]:
          rnd_characters = [
            Character.GODOT,
            Character.FRANZISKA,
            Character.JUDGE,
            Character.LARRY,
            Character.MAYA,
            Character.KARMA,
            Character.PAYNE,
            Character.MAGGEY,
            Character.PEARL,
            Character.LOTTA,
            Character.GUMSHOE,
            Character.GROSSBERG,
          ]
          rnd_character = random.choice(
            list(
              filter(
                lambda character: character not in characters, rnd_characters
              )
            )
          )
          characters[rnd_character] = character
    return characters

def split_str_into_newlines(text: str, max_line_count):
  lines = []
  for line in wrap(text, max_line_count):
    lines.append(line + ' ')
  new_text = '\n'.join(lines)
  return new_text

max_stars_repo_name: shell-bee/ace-attorney-engine

max_issues_repo_name: shell-bee/ace-attorney-engine

max_forks_repo_name: shell-bee/ace-attorney-engine


text:
import numpy
from collections import defaultdict

def main():
    limit = 1000000
    count = defaultdict(int)

    maxi_num = 1
    maxi_val = 0
    for num in range(2, limit + 1):
        num_copy = num
        suma = 0
        while num_copy > 1:
            if num_copy in count:
                suma += count[num_copy]
                break
            else:
                suma += 1

            if (num_copy % 2 == 0):
                num_copy = num_copy // 2
            else:
                num_copy = 3 * num_copy + 1
        count[num] = suma

        if suma > maxi_val:
            maxi_num = num
            maxi_val = suma

    print(maxi_num, maxi_val)

if __name__ == '__main__':
    main()

max_stars_repo_name: jdalzatec/EulerProject

max_issues_repo_name: jdalzatec/EulerProject

max_forks_repo_name: jdalzatec/EulerProject


text:
import tensorflow as tf
import numpy as np

class MyEmbeddingsFromInts(tf.keras.layers.Layer):
    def __init__(self, embeddings, dtype=tf.float32):
        super().__init__()

        unk_emb = np.expand_dims(np.zeros(embeddings.shape[1]), axis=0)
        self._all_embs = tf.constant(np.concatenate((embeddings, unk_emb), axis=0), dtype=dtype)

    def call(self, inputs):
        batch_embs = tf.nn.embedding_lookup(self._all_embs, inputs)
        return batch_embs

max_stars_repo_name: rist-ro/argo

max_issues_repo_name: rist-ro/argo

max_forks_repo_name: rist-ro/argo


text:
# Undergraduate Student: Arturo Burgos
# Professor: João Rodrigo Andrade
# Federal University of Uberlândia - UFU, Fluid Mechanics Laboratory - MFLab, Block 5P, Uberlândia, MG, Brazil

# First exercise: Consider an arbitrary Matrix NxNx3 - vectorized

import numpy as np
import time  # Here I import this python module in order to calculate the precise time that the function works

n = 5000  # The numbers of elements in each matrix dimension

A = np.random.rand(n, n, 3)  # Here I just import numpy to create an arbitrary matrix

print('The matrix is \n', A)
print('\n')
print('The element A[1][1][1] of the matrix is ', A[0][0][0])
print('The element A[1][1][2] of the matrix is ', A[0][0][1])
print('The element A[1][1][3] of the matrix is ', A[0][0][2])  # Here I certified that this number is in the right
# position according to the matrix A
print('\n')

# print(A[:][1][1])

# A[:][:][0] = A[1][:][1] # Here there is a big confusion: Apparently when I set the first row (A[0][...])
# A[:][:][2] = A[:][:][0] # It is the same as if I was doing like A[:][0][...] which doesn't make any sense
# So I already knew that there are two different ways to represent an multi-dimensional array: A[n,n,n] or A[n][n][n]
# And the strange thing is that only the second one work just fine

t_initial = time.time()  # I set here the initial time.

def vectorized(mtx):
    mtx[:, :, 0] = mtx[:, :, 1]
    mtx[:, :, 2] = mtx[:, :, 0]
    return mtx

print('\nThe new matrix is ')
A = vectorized(A)
print('\n')
print(A)
print('\n')
print("--- %s segundos ---" % (time.time() - t_initial))
print('\n')
print('The element A[1][1][1] of the matrix is ', A[0][0][0])
print('The element A[1][1][2] of the matrix is ', A[0][0][1])
print('The element A[1][1][3] of the matrix is ', A[0][0][2])

max_stars_repo_name: bangyen/Comparing-Languages

max_issues_repo_name: bangyen/Comparing-Languages

max_forks_repo_name: bangyen/Comparing-Languages


text:
"""
Author: Luca Soldaini (luca@soldaini.net)

Description: Script to test end-to-end system
    for relation classification/extraction
"""

# built-in modules
import os
import re
import shlex
import typing
import tempfile
import itertools
import subprocess
import collections

# installed modules
import spacy
import numpy as np
from bs4 import BeautifulSoup

# project modules
# None

# constants
INVALID_TAGS = ['text', 'title', 'abstract', 'body', 'html']

def load_abstracts_relations(
        subtask: typing.Union[float, int, str],
        load_test: bool=False
) -> typing.Tuple[list, list, list]:
    """
    Load abstracts, entities, and relations from dataset.

    :param subtask: Subtask to consider. Choose from 1.1, 1.2, or 2.
    :type subtask: typing.Union[float, int, str]

    :return: a tuple cointains:
        -   parsed_texts: a list of dictionaries, each of which containing
            the following keyed values:
                +   id: identifier for the document
                +   text: text of the doucment
        -   parsed_entities: a list of lists of entities in each
            document. Each list of entities associated with a document
            contains the following keyed values:
                +   id: entity identifier
                +   text: textual represenation of the entity
        -   parsed_relations: a list of lists of relations in each
            document. Each list of relations associated with a document
            contains the following keyed values:
                +   ent_a: entity involved in the relationship
                +   ent_b: entity involved in the relationship
                +   is_reverse: if false, relation is ent_a -> ent_b;
                    if true, relation is ent_b -> ent_a
                +   type: type of relationship
    :rtype: typing.Tuple[list, list, list]
    """

    if not load_test:
        # path to dataset containing abstracts and entities
        dataset_path = os.path.join('training-data', str(subtask), 'text.xml')
    else:
        dataset_path = os.path.join('test-data', str(subtask), 'text.xml')

    with open(dataset_path) as f:
        soup = BeautifulSoup(f, 'xml')

    parsed_entities = []
    parsed_texts = []

    # iterate over all titles/abstracts in the input text
    for title_abstract in soup.find_all('text'):

        text_id = title_abstract.attrs['id']

        parsed_passage = ''

        # add a full stop after the title of the paper,
        # so it can be merged with the abstract.
        title_abstract = BeautifulSoup(
            str(title_abstract).replace('</title>', '</title>.'),
            'lxml'
        )

        # remove tags arount title/abstract that we don't care about
        for tag in INVALID_TAGS:
            for match in title_abstract.findAll(tag):
                match.unwrap()

        # throw away all xml syntax, only keep the raw info for
        # easy parsing and removal of entities. we want to go from
        #   ... rhetorical and <entity id="L08-1459.34">syntactic</entity>
        #   <entity id="L08-1459.35">properties</entity> of parentheticals
        #   as well as the <entity id="L08-1459.36">decisions</entity> ...
        # to
        #   ... rhetorical and syntactic properties of parentheticals
        #   as well as the decisions ...
        # plus a list of entities.
        title_abstract = str(title_abstract)

        # remove unwanted spaces and add spaces when not set.
        title_abstract = re.sub(r'<entity ', r' <entity ', title_abstract)
        title_abstract = re.sub(r'>\s*', r'>', title_abstract)
        title_abstract = re.sub(r'\s*</entity>', r'</entity> ', title_abstract)
        title_abstract = re.sub(r'\s+', ' ', title_abstract).strip()
        parsed_entities.append([])

        i = 0   # moving pointer to character in the passsage
        while True:
            # keep looping until all entities have been extracted

            # find the start of the next entity
            next_entity = title_abstract.find('<entity', i)

            if next_entity < 0:
                # if the start is -1, there are no more entities to
                # extract, so add the last remaining bit of text to
                # parsed passage and get out of the while True loop.
                parsed_passage += title_abstract[i:]
                break

            # add the text between the last entity extacted and
            # the new entity to the parsed passage.
            parsed_passage += title_abstract[i:next_entity]

            # determine where the actual entity starts and ends, ignoring
            # the "<entity>" and "</entity>" tags.
            start_entity = title_abstract.find('">', next_entity) + 2
            end_entity = title_abstract.find('</entity>', start_entity)

            # extract the text of the entity
            entity_text = title_abstract[start_entity:end_entity]

            # extract the id of the entity
            entity_id = re.search(
                "\"(.+?)\"", title_abstract[next_entity:start_entity]
            ).groups()[0]

            # calculate where to start to scan next based on the
            # end entity offset and the length on the tag
            i = end_entity + len('</entity>')

            # derive position of the entity in the new parsed passage
            start_entity_in_parsed_passage = len(parsed_passage)
            parsed_passage += entity_text
            end_entity_in_parsed_passage = len(parsed_passage)

            # add entity to list of entitities
            parsed_entities[-1].append({
                'id': entity_id,
                'text': entity_text,
                'start': start_entity_in_parsed_passage,
                'end': end_entity_in_parsed_passage
            })

        # add parsed text to list of texts.
        parsed_texts.append({
            'id': text_id,
            'text': parsed_passage
        })

    if not load_test:
        # path to file containing relations
        relations_path = os.path.join(
            'training-data', str(subtask), 'relations.txt'
        )
    else:
        # path to file containing relations
        relations_path = os.path.join(
            'test-data', str(subtask), 'relations.txt'
        )

    # set up list to group relations by document
    docs_ids = {text['id']: i for i, text in enumerate(parsed_texts)}
    parsed_relations = [[] for _ in docs_ids]

    # get relations out
    # note that no relations are given for subtask 2 test set
    with open(relations_path) as f:
        for ln in f:

            # strip end characters
            ln = ln.strip()

            # skip empty lines
            if not ln:
                continue

            # data format:
            # RELATION_TYPE(<ENTITY>,<ENTITY>)
            # or
            # RELATION_TYPE(<ENTITY>,<ENTITY>,<REVERSE>)

            # if not load_test:
            # separate relation type from entities
            rel_type, rel_data = ln.strip(')').split('(')
            # else:
            # rel_type = None
            # rel_data = ln.strip(')').strip('(')

            # parse entities, reverse if avaliable
            try:
                ent_a, ent_b, is_reverse = rel_data.split(',')
            except ValueError:
                ent_a, ent_b = rel_data.split(',')
                is_reverse = False

            # use doc id to determine the position in parsed_relations list
            doc_id = ent_a.split('.')[0]

            # casting to prevent warning in PyCharm
            doc_pos = int(docs_ids[doc_id])

            parsed_relations[doc_pos].append({
                'type': rel_type,
                'ent_a': ent_a,
                'ent_b': ent_b,
                'is_reverse': is_reverse,
            })

    return parsed_texts, parsed_entities, parsed_relations

def split_dataset_into_sentences(
        texts: list,
        entities: list,
        relations: list,
        include_negative_samples=True
) -> list:
    """
    Reformat training data to have positive (and negative samples, if needed),
    and to be organized per sentence.

    :param texts: text of input abstracts
    :type texts: list

    :param entities: list of entites in document
    :type entities: list

    :param relations: list of relations in documents
    :type relations: list

    :param include_negative_samples: if true, negative samples
        are included in the set
    :type include_negative_samples: bool

    :return: list of positive (and negative samples)
        for relation classification
    :rtype: list
    """

    nlp = spacy.load('en')

    samples = []

    for text, doc_ents, doc_rels in zip(texts, entities, relations):

        sentences = [sent for sent in nlp(text['text']).sents]

        # map the character start position of tokens in
        # passage to each sentence / position in sentence
        token_starts = {
            pos: (sent_id, sent_pos)
            for pos, sent_id, sent_pos in itertools.chain(*(
                [(token.idx, s, t) for t, token in enumerate(sent)]
                for s, sent in enumerate(sentences)
            ))
        }

        # map the character end position of tokens in
        # passage to each sentence / position in sentence
        token_ends = {
            pos: (sent_id, sent_pos)
            for pos, sent_id, sent_pos in itertools.chain(*(
                [
                    (token.idx + len(token), s, t)
                    for t, token in enumerate(sent)
                ]
                for s, sent in enumerate(sentences)
            ))
        }

        # reorganize relations in dictionary for easier use
        dict_doc_rels = {}
        for rel in doc_rels:
            id_ent_a = rel['ent_a'].split('.')
            id_ent_a = id_ent_a[0], int(id_ent_a[1])

            id_ent_b = rel['ent_b'].split('.')
            id_ent_b = id_ent_b[0], int(id_ent_b[1])

            dict_doc_rels.setdefault(
                id_ent_a, {}
            )[id_ent_b] = rel['type'], rel['is_reverse']

        # reorganize entites in dictionary for easier use
        sents_ents = collections.defaultdict(list)
        for ent in doc_ents:
            ent_sent_id, ent_sent_pos_start = token_starts[ent['start']]
            _, ent_sent_pos_end = token_ends[ent['end']]

            # offset by 1 for easier indexing using
            # list slicing
            ent_sent_pos_end += 1

            ent_id = ent['id'].split('.')
            ent_id = ent_id[0], int(ent_id[1])

            sents_ents[ent_sent_id].append({
                'id': ent_id,
                'text': ent['text'],
                'sentence': ent_sent_id,
                'sent_start': ent_sent_pos_start,
                'sent_end': ent_sent_pos_end,
                'start': ent['start'],
                'end': ent['end']
            })

        for s, sent in enumerate(sentences):

            tokens = [token.text.lower() for token in sent]

            for ent_a, ent_b in itertools.combinations(sents_ents[s], 2):

                # swap if entity a occurs after entity b
                if ent_a['id'][1] > ent_b['id'][1]:
                    ent_a, ent_b = ent_b, ent_a

                rel_info = \
                    dict_doc_rels.get(ent_a['id'], {}).get(ent_b['id'], None)

                # relation exists between the two entities
                if rel_info:
                    rel_type, rel_reverse = rel_info
                    rel_reverse = 1 if rel_reverse else 0
                else:
                    rel_type, rel_reverse = 'NONE', 0

                if not include_negative_samples and rel_type == 'NONE':
                    # skip here if negative samples are not needed
                    continue

                samples.append({
                    'id': text['id'],
                    'tokens': tokens,
                    'spacy': sent,
                    'ent_a_start': ent_a['sent_start'],
                    'ent_a': '{}.{}'.format(*ent_a['id']),
                    'ent_a_end': ent_a['sent_end'],
                    'ent_b_start': ent_b['sent_start'],
                    'ent_b': '{}.{}'.format(*ent_b['id']),
                    'ent_b_end': ent_b['sent_end'],
                    'type': rel_type,
                    'is_reverse': rel_reverse,
                })

    print('[info] {:,} total samples'.format(len(samples)))

    return samples

def get_dependencies_map(dataset: tuple) -> dict:
    """Parse training """
    texts, _, _ = dataset
    nlp = spacy.load('en')

    dependencies_map = {'<UNK>': 0}

    for text in texts:
        parsed = nlp(text['text'])
        for token in parsed:
            dependencies_map.setdefault(token.dep_, len(dependencies_map))

    return dependencies_map

def get_part_of_speech_map(dataset: tuple) -> dict:
    """Parse training """
    texts, _, _ = dataset
    nlp = spacy.load('en')

    pos_map = {'<UNK>': 0}

    for text in texts:
        parsed = nlp(text['text'])
        for token in parsed:
            pos_map.setdefault(token.pos_, len(pos_map))

    return pos_map

def get_distribution_ent_length(sentences: list) -> typing.Tuple[float, float]:
    """Parse training """

    lengths = []

    for sent in sentences:
        lengths.append(sent['ent_a_end'] - sent['ent_a_start'])
        lengths.append(sent['ent_b_end'] - sent['ent_b_start'])

    return float(np.average(lengths)), float(np.std(lengths))

def split_train_test(
        subtask: typing.Union[float, int, str],
        texts: list,
        entities: list,
        relations: list
) -> typing.Tuple[tuple, tuple]:
    """
    Split list of texts, entities, and relations into
    training and test datasets.

    :param subtask: Subtask to consider. Choose from 1.1, 1.2, or 2.
    :type subtask: typing.Union[float, int, str]

    :param texts: text of input abstracts
    :type texts: list

    :param entities: list of entites in document
    :type entities: list

    :param relations: list of relations in documents
    :type relations: list

    :return: A tuple of (texts, entities, relations) tuples split
        between training and text data
    :rtype typing.Tuple[tuple, tuple]
    """
    texts_train, texts_test = [], []
    entities_train, entities_test = [], []
    relations_train, relations_test = [], []

    # load the
    test_doc_list_path = os.path.join(
        'training-data', str(subtask), 'test.txt'
    )
    with open(test_doc_list_path) as f:
        test_set = {ln.strip() for ln in f}

    for doc_text, doc_entities, doc_relations in\
            zip(texts, entities, relations):
        if doc_text['id'] in test_set:
            texts_test.append(doc_text)
            entities_test.append(doc_entities)
            relations_test.append(doc_relations)
        else:
            texts_train.append(doc_text)
            entities_train.append(doc_entities)
            relations_train.append(doc_relations)

    return (
        (texts_train, entities_train, relations_train),
        (texts_test, entities_test, relations_test)
    )

def split_train_test_sentences(
        subtask: typing.Union[float, int, str],
        samples: list,
) -> typing.Tuple[list, list]:
    """
    Split list of texts, entities, and relations into
    training and test datasets.

    :param subtask: Subtask to consider. Choose from 1.1, 1.2, or 2.
    :type subtask: typing.Union[float, int, str]

    :param samples: list of positive (and negative samples)
        for relation classification
    :type samples: list

    :return: Two lists of training and test samples.
    :rtype typing.Tuple[tuple, tuple]
    """

    test_doc_list_path = os.path.join(
        'training-data', str(subtask), 'test.txt'
    )
    with open(test_doc_list_path) as f:
        test_set = {ln.strip() for ln in f}

    train_samples, test_samples = [], []
    for sample in samples:
        if sample['id'] in test_set:
            if sample['type'] == 'NONE':
                continue
            test_samples.append(sample)
        else:
            train_samples.append(sample)

    return train_samples, test_samples

def _format_output(relations: list) -> str:
    """
    Convert a file with relations to a string that can be
    evaluated by eval.pl script.

    :param relations: relations file to parse
    :return:
    """

    if len(relations) == 0:
        return ''

    if type(relations[0]) is list:
        # relations are nested, so let's un-nest them
        relations = itertools.chain(*relations)

    output = []
    for rel in relations:

        if rel['type'] == 'NONE':
            continue

        rel_repr = (
            '{}({},{},REVERSE)' if rel['is_reverse'] else '{}({},{})'
        ).format(rel['type'], rel['ent_a'], rel['ent_b'])

        output.append(rel_repr)

    return '\n'.join(output)

def write_predictions_to_file(predictions: list, path: str):
    """
       Write predictions to file. Predictions should have the same
       format used in the training data, i.e. each relation should
       be a dictionary with the following keys:
           -   ent_a: entity involved in the relationship
           -   ent_b: entity involved in the relationship
           -   is_reverse: if false, relation is ent_a -> ent_b;
               if true, relation is ent_b -> ent_a
           -   type: type of relationship

       :param predictions: predicted labels by a given relation
           extraction method.
       :type predictions: list

       :param path: destination of predictions
       :type path: str

       """

    with open(path, 'w') as f_preds:
        f_preds.write(_format_output(predictions))

def evaluate(labels: list, predictions: list) -> str:
    """
    Evaluate predictions. Predictions should have the same format used
    in the training data, i.e. each relation should be a dictionary
    with the following keys:
        -   ent_a: entity involved in the relationship
        -   ent_b: entity involved in the relationship
        -   is_reverse: if false, relation is ent_a -> ent_b;
            if true, relation is ent_b -> ent_a
        -   type: type of relationship

    :param labels: labels from documents in the test set
    :type labels: list

    :param predictions: predicted labels by a given relation
        extraction method.
    :type predictions: list

    :return: the output of evaluation tool "eval.pl"
    :rtype: str
    """

    with tempfile.NamedTemporaryFile('w', delete=False) as f_preds, \
            tempfile.NamedTemporaryFile('w', delete=False) as f_labels:

        f_preds.write(_format_output(predictions))
        f_labels.write(_format_output(labels))

        pred_path, labels_path = f_preds.name, f_labels.name

    cmd = shlex.split('perl eval.pl {} {}'.format(pred_path, labels_path))
    proc = subprocess.Popen(
        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
    raw_output = proc.communicate()

    stdout, stderr = map(lambda e: e.decode('utf-8'), raw_output)
    os.remove(pred_path)
    os.remove(labels_path)

    if stderr.strip():
        raise RuntimeError(stderr.strip())

    return stdout


text:
import numpy as np
import tensorflow as tf

def _xavier_init(fan_in, fan_out, constant=1):
    """
    Xavier initialization of network weights.
    For some explanations see:
        - http://andyljones.tumblr.com/post/110998971763/an-explanation-of-xavier-initialization
        - http://jmlr.org/proceedings/papers/v9/glorot10a/glorot10a.pdf
        - https://stackoverflow.com/questions/33640581/how-to-do-xavier-initialization-on-tensorflow
    """
    low = -constant * np.sqrt(6.0 / (fan_in + fan_out))
    high = constant * np.sqrt(6.0 / (fan_in + fan_out))
    return tf.random_uniform((fan_in, fan_out),
                             minval=low, maxval=high,
                             dtype=tf.float32)

class SimpleAutoencoder(object):
    """
    Autoencoder implementation based on inputs from https://jmetzen.github.io/2015-11-27/vae.html.
    See the project README on how to use it.
    """

    def __init__(self,
                 network_architecture,
                 transfer_fct=tf.nn.tanh,
                 batch_size=100,
                 weight_regularization=0.0,
                 bias_regularization=0.0,
                 tied=False):

        self.network_architecture = network_architecture
        self.transfer_fct = transfer_fct
        self.batch_size = batch_size
        self.weight_regularization = weight_regularization
        self.bias_regularization = bias_regularization
        self.tied = tied
        self._encoder_layers = []
        self._decoder_layers = []
        self._recording = {'epoch': 0,
                           'learning_rate_schedule': dict(),
                           'batch_cost': []}

        # tf Graph input
        self.learning_rate = tf.placeholder(tf.float32, shape=[])
        self.x = tf.placeholder(tf.float32, [None, network_architecture["n_input"]])
        self._create_network()

        # Define loss function based variational upper-bound and
        # corresponding optimizer
        self._create_loss_optimizer()

        # Initializing the tensor flow variables
        init = tf.initialize_all_variables()

        # Launch the session
        num_cores = 6
        self.sess = tf.InteractiveSession(config=tf.ConfigProto(device_count={'CPU': num_cores},
                                                                inter_op_parallelism_threads=num_cores,
                                                                intra_op_parallelism_threads=num_cores))
        self.sess.run(init)

    def _create_network(self):
        # Initialize autoencoder network weights and biases
        self.network_weights = self._initialize_weights(self.network_architecture, self.tied)

        if self.tied:
            self.x_encoded, self.x_decoded = self._tied_encoder_network(self.network_weights["weights_encoder"],
                                                                        self.network_weights["biases_encoder"],
                                                                        self.network_weights["biases_decoder"])

        else:

            assert 'decoder' in self.network_architecture.keys(), "Need to provide decoder network layer configuration for untied encoder."

            self.x_encoded = self._encoder_network(self.network_weights["weights_encoder"],
                                                   self.network_weights["biases_encoder"])

            self.x_decoded = self._decoder_network(self.network_weights["weights_decoder"],
                                                   self.network_weights["biases_decoder"])

    @staticmethod
    def _create_matrix_and_bias_sizes(n_input, lst_of_encoder_layer_sizes, n_compressed):
        matrix_dims = zip([n_input] + lst_of_encoder_layer_sizes, lst_of_encoder_layer_sizes + [n_compressed])
        bias_dims = [tup[1] for tup in matrix_dims]
        return matrix_dims, bias_dims

    @staticmethod
    def _create_layers(component, lst_of_component_layer_sizes, n_input, n_compressed):

        all_weights = dict()
        if len(lst_of_component_layer_sizes) > 0:

            matrix_dims, bias_dims = SimpleAutoencoder._create_matrix_and_bias_sizes(n_input,
                                                                                lst_of_component_layer_sizes,
                                                                                n_compressed)
            all_weights['weights_{}'.format(component)] = dict()
            all_weights['biases_{}'.format(component)] = dict()

            for idx, dim in enumerate(matrix_dims):
                print "{} layer {}, dimensionality {} -> {}".format(component, idx + 1, dim[0], dim[1])
                all_weights['weights_{}'.format(component)].update(
                    {'h{}'.format(idx + 1): tf.Variable(_xavier_init(dim[0], dim[1]))})

                all_weights['biases_{}'.format(component)].update(
                    {'b{}'.format(idx + 1): tf.Variable(tf.zeros([bias_dims[idx]], dtype=tf.float32))})

            return all_weights

        else:
            raise AttributeError("need list of {} layer sizes".format(component))

    @staticmethod
    def _initialize_weights(architecture_dict, tied):

        all_weights = dict()

        n_input = architecture_dict.get('n_input')
        n_compressed = architecture_dict.get('n_compressed')
        all_weights.update(SimpleAutoencoder._create_layers('encoder', architecture_dict.get('encoder'), n_input, n_compressed))

        if tied:
            lst_of_dims = architecture_dict.get('encoder')

            # List `reverse` manipulates IN-PLACE and returns None. Hence we cannot put this inline :(
            lst_of_dims.reverse()
            all_weights.update(SimpleAutoencoder._create_layers('decoder', lst_of_dims, n_compressed, n_input))

        else:
            assert 'decoder' in architecture_dict.keys(), "Need to provide decoder network layer configuration for untied encoder."

            all_weights.update(SimpleAutoencoder._create_layers('decoder', architecture_dict.get('decoder'), n_compressed, n_input))

        return all_weights

    def _tied_encoder_network(self, weights, biases_encoder, biases_decoder):

        layers = []
        for idx in range(len(weights)):
            if idx == 0:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(self.x, weights['h{}'.format(idx + 1)]), biases_encoder['b{}'.format(idx + 1)])))
            else:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(layers[idx - 1], weights['h{}'.format(idx + 1)]), biases_encoder['b{}'.format(idx + 1)])))

        for idx in range(len(weights)):
            if idx == 0:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(layers[-1], weights['h{}'.format(len(weights) - idx)], transpose_b=True), biases_decoder['b{}'.format(idx + 1)])))
            else:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(layers[len(weights) + idx -1], weights['h{}'.format(len(weights) - idx)], transpose_b=True),
                           biases_decoder['b{}'.format(idx + 1)])))

        self._encoder_layers = layers[:len(weights)]
        self._decoder_layers = layers[len(weights):]
        return layers[len(weights) - 1], layers[-1]

    def _encoder_network(self, weights, biases):

        layers = []
        for idx in range(len(weights)):
            if idx == 0:
                layers.append(self.transfer_fct(tf.add(tf.matmul(self.x, weights['h{}'.format(idx + 1)]), biases['b{}'.format(idx +1)])))
            else:
                layers.append(self.transfer_fct(tf.add(tf.matmul(layers[idx-1], weights['h{}'.format(idx + 1)]), biases['b{}'.format(idx +1)])))

        self._encoder_layers = layers
        return layers[-1]

    def _decoder_network(self, weights, biases):

        layers = []
        for idx in range(len(weights)):
            if idx == 0:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(self.x_encoded, weights['h{}'.format(idx + 1)]), biases['b{}'.format(idx + 1)])))
            else:
                layers.append(self.transfer_fct(
                    tf.add(tf.matmul(layers[idx - 1], weights['h{}'.format(idx + 1)]), biases['b{}'.format(idx + 1)])))

        self._decoder_layers = layers
        return layers[-1]

    def _create_loss_optimizer(self):

        # The reconstruction loss
        reconstr_loss = tf.reduce_sum(tf.squared_difference(self.x, self.x_decoded))

        # Weight matrix regularization loss
        weight_reg = 0
        for val in self.network_weights['weights_encoder'].itervalues():
            weight_reg += tf.reduce_sum(tf.square(val))

        for val in self.network_weights['weights_decoder'].itervalues():
            weight_reg += tf.reduce_sum(tf.square(val))

        # Bias vector regularization loss
        bias_reg = 0
        for val in self.network_weights['biases_encoder'].itervalues():
            bias_reg += tf.reduce_sum(tf.square(val))

        for val in self.network_weights['biases_decoder'].itervalues():
            bias_reg += tf.reduce_sum(tf.square(val))

        self.cost = tf.reduce_mean(reconstr_loss) + (self.weight_regularization * weight_reg + self.bias_regularization * bias_reg) / (2 * self.batch_size)

        self.optimizer = tf.train.GradientDescentOptimizer(learning_rate=self.learning_rate).minimize(self.cost)

    def partial_fit(self, X, learning_rate=None):
        """Train model based on mini-batch of input data.

        Return cost of mini-batch.
        """
        if not learning_rate:
            self.learning_rate = 0.001

        opt, cost = self.sess.run((self.optimizer, self.cost),
                                  feed_dict={self.x: X, self.learning_rate: learning_rate})
        return cost

    def encode(self, X):
        """Transform data by mapping it into the latent space."""
        return self.sess.run(self.x_encoded, feed_dict={self.x: X})

    def decode(self, X):
        return self.sess.run(self.x_decoded, feed_dict={self.x: X})

    def _monitor_layer(self, X, layer_index, network='encoder'):

        if network == 'encoder':
            layer_to_run = self._encoder_layers[layer_index]

        elif network == 'decoder':
            layer_to_run = self._decoder_layers[layer_index]

        else:
            raise AttributeError("network '{}' does not exist".format(network))

        return self.sess.run(layer_to_run, feed_dict={self.x: X})

    def _update_learning_rate(self, dct, epoch):

        epoch_key = max(k for k in dct if k <= epoch)
        if self._recording['epoch'] <= 1:
            self._current_lr = dct[epoch_key]
            self._recording['learning_rate_schedule'].update({self._recording['epoch'] - 1: self._current_lr})

        if dct[epoch_key] != self._current_lr:
            self._current_lr = dct[epoch_key]
            self._recording['learning_rate_schedule'].update({self._recording['epoch'] - 1: self._current_lr})

        return self._current_lr

    def train(self, X, n_epochs, learning_rate=0.0001, display_step=10):

        data = DataSet(X, self.batch_size)

        for epoch in range(n_epochs):

            self._recording['epoch'] += 1

            if isinstance(learning_rate, dict):
                lr = self._update_learning_rate(learning_rate, epoch)
            else:
                lr = learning_rate

            data.reset_counter()
            costs = []

            while data.has_next():
                batch_xs = data.next_batch()
                costs.append(self.partial_fit(batch_xs, lr))

            self._recording['batch_cost'].append(np.mean(costs))

            # Display logs per epoch step
            if (epoch+1) % display_step == 0:
                print "Epoch:", '{0:04d} / {1:04d}'.format(epoch + 1, n_epochs), \
                    "cost=", "{:.9f}".format(self._recording['batch_cost'][-1])

class DataSet(object):

    def __init__(self, data, batch_size = 100):
        self.data = data
        self.n_samples = self.data.shape[0]
        self.n_dimensions = self.data.shape[1]
        self.batch_size = batch_size
        self.total_batches = int(self.n_samples / self.batch_size)
        self.current_batch = 0

    def next_batch(self):
        batch_lower_idx = self.current_batch * self.batch_size
        batch_upper_idx = (self.current_batch + 1) * self.batch_size

        self.current_batch += 1

        return self.data[batch_lower_idx:batch_upper_idx, :]

    def has_next(self):
        return self.current_batch < self.total_batches

    def reset_counter(self):
        # Shuffle data set on re-initialization
        # Note that shuffle does this IN-PLACE. Sad :(
        np.random.shuffle(self.data)
        self.current_batch = 0

max_stars_repo_name: jotterbach/dstk

max_issues_repo_name: jotterbach/dstk

max_forks_repo_name: jotterbach/dstk


text:
from collections import defaultdict

from scipy.stats import stats
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline

from lib2vec.corpus_structure import DataHandler
from lib2vec.vectorization_utils import Vectorization
import sklearn as sk
from sklearn.metrics import f1_score
from sklearn.metrics import make_scorer, accuracy_score, precision_score, recall_score, f1_score
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from statsmodels.stats.contingency_tables import mcnemar

def cross_val_scores_weighted(model, X, y, weights, cv=5, metrics=[sk.metrics.accuracy_score]):
    kf = sk.model_selection.KFold(n_splits=cv)
    kf.get_n_splits(X)
    scores = [[] for metric in metrics]
    for train_index, test_index in kf.split(X):
        model_clone = sk.base.clone(model)
        print(train_index, test_index)
        X_train, X_test = X[train_index], X[test_index]
        y_train, y_test = y[train_index], y[test_index]
        weights_train, weights_test = weights[train_index], weights[test_index]
        model_clone.fit(X_train,y_train,sample_weight=weights_train)
        y_pred = model_clone.predict(X_test)
        for i, metric in enumerate(metrics):
            score = metric(y_test, y_pred, sample_weight = weights_test)
            scores[i].append(score)
    return scores

def chi_square_test(input1, input2):
    truth1 = [0.5 for e in input1]
    truth2 = [0.5 for e in input2]
    print('---')
    print(stats.chisquare(input1, truth1))
    print(stats.chisquare(input2, truth2))
    print('---')

def mcnemar_sig_text(table):
    # define contingency table
    # calculate mcnemar test
    exact = False
    correction = True
    for line in table:
        for e in line:
            if e < 25:
                exact = True
                correction = False
    result = mcnemar(table, exact=exact, correction=correction)
    # summarize the finding
    print('statistic=%.3f, p-value=%.3f' % (result.statistic, result.pvalue))
    # interpret the p-value
    alpha = 0.05
    if result.pvalue > alpha:
        print(f'Classifiers have same proportions of errors (fail to reject H0), {result.pvalue}')
    else:
        print(f'Classifiers have different proportions of errors (reject H0), {result.pvalue}')

def success_prediction_task(data_set_name: str, success_dict, vector_names):
    result_tuples = []
    correctness_table = defaultdict(list)
    for vectorization_algorithm in vector_names:
        majority_class = None
        if vectorization_algorithm == "majority_class":
            majority_class = vectorization_algorithm
            vectorization_algorithm = "doc2vec"

        vec_path = Vectorization.build_vec_file_name("all",
                                                     "no_limit",
                                                     data_set_name,
                                                     "no_filter",
                                                     vectorization_algorithm,
                                                     "real",
                                                     allow_combination=True)
        vectors, summation_method = Vectorization.my_load_doc2vec_format(vec_path)

        x = []
        y = []
        doc_ids = []
        k_fold_cross_val = None
        for doctag in vectors.docvecs.doctags:
            try:
                if summation_method and f"_{summation_method}" in str(doctag):
                    x.append(vectors.docvecs[doctag])
                    # success = success_dict[doctag.replace(f"_{summation_method}", "")]
                    # if success == "failure":
                    #     success = 0
                    # else:
                    #     success = 1
                    # y.append(success)
                    y.append(0 if success_dict[doctag.replace(f"_{summation_method}", "")] == "failure" else 1)
                    doc_ids.append(doctag.replace(f"_{summation_method}", ""))

                elif not summation_method and str(doctag)[-1].isdigit():
                    doc_splitted = doctag.split("_")
                    if len(doc_splitted) > 1 and doc_splitted[-1][-1].isdigit() and doc_splitted[-2][-1].isdigit():
                        pass
                    else:
                        x.append(vectors.docvecs[doctag])
                        y.append(0 if success_dict[doctag] == "failure" else 1)
                        doc_ids.append(doctag)
                else:
                    pass
            except KeyError:
                pass

        counter = defaultdict(lambda: 0)
        print(len(y))
        for truth_val in y:
            counter[truth_val] += 1
        # print(len(doc_ids), len(y))
        # print(doc_ids)

        # print(y)
        classifiers = [
            # RandomForestClassifier(n_estimators=100, max_depth=2, random_state=0),
            sk.svm.LinearSVC(max_iter=10000, class_weight="balanced", dual=False, random_state=42, C=1.5),
            sk.svm.SVC(max_iter=10000, class_weight="balanced", kernel="rbf", random_state=42, C=0.5),
            sk.svm.NuSVC(max_iter=10000, class_weight="balanced", kernel="rbf", gamma="scale", random_state=42, nu=0.5),
            # MLPClassifier(solver='adam', alpha=1e-5, hidden_layer_sizes=(10, 4), random_state=42, max_iter=10000),
            # LogisticRegression(random_state=0, solver='lbfgs', multi_class='ovr', max_iter=1000),
        ]

        # classifiers = [make_pipeline(StandardScaler(), classifier) for classifier in classifiers]
        for classifier in classifiers:
            pipeline_classifier = make_pipeline(StandardScaler(), classifier)
            if k_fold_cross_val:
                weights = [counter[pred] for pred in y]
                # results = cross_val_scores_weighted(classifier, x, y, weights, cv=k_fold_cross_val,
                #                                     metrics=[f1_score, precision_score,
                #                                              recall_score, f1_score])

                scoring = {
                    'accuracy': make_scorer(accuracy_score),
                    'precision': make_scorer(precision_score),
                    'recall': make_scorer(recall_score),
                    'f1_score': make_scorer(f1_score)
                }
                kfold = sk.model_selection.KFold(n_splits=k_fold_cross_val, random_state=42, shuffle=True)
                results = sk.model_selection.cross_validate(estimator=pipeline_classifier,
                                                            X=x,
                                                            y=y,
                                                            cv=kfold,
                                                            scoring=scoring)

                result_tuples.append((classifier.__class__.__name__,
                                      vectorization_algorithm,
                                      np.mean(results["test_f1_score"]),
                                      np.mean(results["test_precision"]),
                                      np.mean(results["test_recall"]),
                                      np.mean(results["test_accuracy"])))
            else:
                nr_iterations = 100
                f1s = []
                for i in range(nr_iterations):
                    x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.30, random_state=i)
                    pipeline_classifier.fit(x_train, y_train)
                    y_pred = pipeline_classifier.predict(x_test)

                    if majority_class:
                        y_pred = [1 for y in y_pred]
                        vectorization_algorithm = majority_class

                    # weights = [counter[pred] for pred in y_test]
                    correctness_table[f"{vectorization_algorithm}-{classifier.__class__.__name__}"]\
                        .extend([pred == truth for pred, truth in zip(y_pred, y_test)])
                    f1s.append(f1_score(y_test, y_pred, average="weighted", pos_label=None))
                    # print(weights)
                    # print(vectorization_algorithm, round(classifier.score(x_test, y_test), 4))

                if len(f1s) == 0:
                    f1s.append(0)
                result_tuples.append((classifier.__class__.__name__,
                                      vectorization_algorithm,
                                      np.mean(f1s),
                                      # precision_score(y_test, y_pred, average="weighted"),
                                      # recall_score(y_test, y_pred,  average="weighted"),
                                      # accuracy_score(y_test, y_pred)
                                      )
                                     )

    print(correctness_table.keys())
    classifiers = [
        "LinearSVC",
        "SVC",
        "NuSVC"
    ]
    for cl in classifiers:
        algo1 = "xlm_pt"
        algo2 = "book2vec_concat"
        true_true = 0
        true_false = 0
        false_true = 0
        false_false = 0
        for e1, e2 in zip(correctness_table[f"{algo1}-{cl}"], correctness_table[f"{algo2}-{cl}"]):
            if e1 and e2:
                true_true += 1
            elif e1 and not e2:
                true_false += 1
            elif not e1 and e2:
                false_true += 1
            elif not e1 and not e2:
                false_false += 1
            else:
                pass
        table = [[true_true, true_false],
                 [false_true, false_false]]
        print(table)
        print(cl)
        mcnemar_sig_text(table)

    df = pd.DataFrame(result_tuples, columns=["Classifier", "Algorithm", "Weighted F1",
                                              # "Precision", "Recall",
                                              # "Accuracy"
                                              ])
    print(df)
    df = df.pivot(index='Algorithm', columns='Classifier', values='Weighted F1')
    df.to_csv("../results/book_success_prediction/eval_scores.csv", index=True)
    print(df)
    print(df.to_latex())

    # df = df.round(4)

def genre_prediction_task(data_set_name: str, genre_dict, vector_names):
    result_tuples = []
    for vectorization_algorithm in vector_names:

        vec_path = Vectorization.build_vec_file_name("all",
                                                     "no_limit",
                                                     data_set_name,
                                                     "no_filter",
                                                     vectorization_algorithm,
                                                     "real",
                                                     allow_combination=True)
        vectors, summation_method = Vectorization.my_load_doc2vec_format(vec_path)

        x = []
        y = []

        k_fold_cross_val = None

        for doctag in vectors.docvecs.doctags:
            if summation_method and f"_{summation_method}" in str(doctag):
                x.append(vectors.docvecs[doctag])
                # success = success_dict[doctag.replace(f"_{summation_method}", "")]
                # if success == "failure":
                #     success = 0
                # else:
                #     success = 1
                # y.append(success)
                y.append(genre_dict[doctag.replace(f"_{summation_method}", "")])

            else:
                if str(doctag)[-1].isdigit():
                    x.append(vectors.docvecs[doctag])
                    y.append(genre_dict[doctag])

        # print(y)
        classifiers = [
            # RandomForestClassifier(n_estimators=100, max_depth=2, random_state=0),
            # sk.svm.LinearSVC(max_iter=10000, class_weight="balanced", dual=False, random_state=42),
            sk.svm.LinearSVC(max_iter=20000, class_weight="balanced", dual=True, random_state=42),
            # sk.svm.LinearSVC(max_iter=10000, class_weight="balanced", dual=True, random_state=42, C=5),
            # sk.svm.LinearSVC(max_iter=10000, dual=False, random_state=42),
            # sk.svm.LinearSVC(max_iter=10000, dual=True, random_state=42),
            # MLPClassifier(solver='adam', alpha=1e-5, hidden_layer_sizes=(5, 2), random_state=42, max_iter=10000),
            # LogisticRegression(random_state=0, solver='lbfgs', multi_class='ovr', max_iter=1000)
        ]
        for classifier in classifiers:
            if k_fold_cross_val:
                scoring = {
                    'accuracy': make_scorer(accuracy_score),
                    'precision': make_scorer(precision_score),
                    'recall': make_scorer(recall_score),
                    'f1_score': make_scorer(f1_score)
                }

                kfold = sk.model_selection.KFold(n_splits=k_fold_cross_val, random_state=42, shuffle=True)

                results = sk.model_selection.cross_validate(estimator=classifier,
                                                            X=x,
                                                            y=y,
                                                            cv=kfold,
                                                            scoring=scoring)

                result_tuples.append((classifier.__class__.__name__,
                                      vectorization_algorithm,
                                      np.mean(results["test_f1_score"]),
                                      np.mean(results["test_precision"]),
                                      np.mean(results["test_recall"]),
                                      np.mean(results["test_accuracy"])))
            else:
                x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.30, random_state=42)
                classifier.fit(x_train, y_train)
                y_pred = classifier.predict(x_test)
                # print(z)
                # print(vectorization_algorithm, round(classifier.score(x_test, y_test), 4))
                result_tuples.append((classifier.__class__.__name__,
                                      vectorization_algorithm,
                                      f1_score(y_test, y_pred, average='weighted'),
                                      precision_score(y_test, y_pred, average='weighted'),
                                      recall_score(y_test, y_pred, average='weighted'),
                                      accuracy_score(y_test, y_pred)))
    df = pd.DataFrame(result_tuples, columns=["Classifier", "Algorithm", "F1", "Precision", "Recall", "Accuracy"])
    df = df.pivot(index='Algorithm', columns=['Classifier'], values='F1')
    df.to_csv("results/book_success_prediction/genre_eval_scores.csv", index=True)
    print(df)
    print(df.to_latex())

if __name__ == '__main__':
    c = DataHandler.load_maharjan_goodreads()
    algorithms = [
        # # "majority_class",
        # "avg_wv2doc",
        # "bow",
        # "avg_wv2doc_restrict10000",
        # "doc2vec",
        # "doc2vec_chunk",
        # "psif",
        # "bert_pt",
        # "bert_sentence_based_1000_pt",
        # "roberta_pt",
        # "roberta_sentence_based_1000_pt",
        # "xlm_pt",
        # "xlm_sentence_based_1000_pt",
        # "doc2vec_chunk",
        # "book2vec",
        # "book2vec_avg",
        # "book2vec_net_only_concat",
        # "book2vec_concat",
        # "book2vec_pca",
        # "book2vec_auto",
        # "book2vec_o_raw",
        # "book2vec_o_time",
        # "book2vec_o_plot",
        # "book2vec_o_loc",
        # "book2vec_o_atm",
        # "book2vec_o_sty",
        # "book2vec_wo_raw_concat",
        # "book2vec_net_concat",
        # "book2vec_dbow",
        # "book2vec_dbow_avg",
        # "book2vec_dbow_concat",
        # "book2vec_dbow_pca",
        # "book2vec_dbow_auto",
        "bert_sentence_based_1000_pt",
        "roberta_sentence_based_1000_pt",
        "xlm_sentence_based_1000_pt",

        # "book2vec_dbow_concat",
        # "bookwvec_window"
        # "book2vec_adv",
        # "book2vec_adv_concat",
    ]
    # EvaluationUtils.build_corpora()
    # EvaluationUtils.train_vecs()

    success_prediction_task("goodreads_genres", c.success_dict, algorithms)

    # genre_dict = {doc_id: document.genres for doc_id, document in c.documents.items()}
    #
    # genre_names = set([genre for doc_id, genre in genre_dict.items()])
    # genre_replace = {genre: i for i, genre in enumerate(genre_names)}
    # genre_dict = {doc_id: genre_replace[genre] for doc_id, genre in genre_dict.items()}
    #
    # genre_prediction_task("goodreads_genres", genre_dict, algorithms)

    df = pd.read_csv("../results/book_success_prediction/eval_scores.csv", index_col="Algorithm")

    # df = df.pivot(index='Algorithm', columns=['Classifier'], values='F1')
    print(df)
    print(df.columns)
    df = df[["NuSVC"]]
    df = df.round(4)
    print(df)
    print(df.to_latex())

max_stars_repo_name: LasLitz/ma-doc-embeddings

max_issues_repo_name: LasLitz/ma-doc-embeddings

max_forks_repo_name: LasLitz/ma-doc-embeddings

reason: import numpy,from scipy,from statsmodels


text:
"""Tests for trace_distance."""
import numpy as np

from toqito.state_metrics import trace_distance
from toqito.states import basis

def test_trace_distance_same_state():
    r"""Test that: :math:`T(\rho, \sigma) = 0` iff `\rho = \sigma`."""
    e_0, e_1 = basis(2, 0), basis(2, 1)
    e_00 = np.kron(e_0, e_0)
    e_11 = np.kron(e_1, e_1)

    u_vec = 1 / np.sqrt(2) * (e_00 + e_11)
    rho = u_vec * u_vec.conj().T
    sigma = rho

    res = trace_distance(rho, sigma)

    np.testing.assert_equal(np.isclose(res, 0), True)

def test_trace_distance_non_density_matrix():
    r"""Test trace distance on non-density matrix."""
    rho = np.array([[1, 2], [3, 4]])
    sigma = np.array([[5, 6], [7, 8]])

    with np.testing.assert_raises(ValueError):
        trace_distance(rho, sigma)

if __name__ == "__main__":
    np.testing.run_module_suite()

max_stars_repo_name: paniash/toqito

max_issues_repo_name: paniash/toqito

max_forks_repo_name: paniash/toqito