D-Net

package eu.dnetlib.data.transform;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.graph.model.DNGFDecoder;

import eu.dnetlib.data.graph.model.DNGFRowKeyDecoder;

import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;

import eu.dnetlib.data.proto.DNGFProtos.DNGF;

import eu.dnetlib.data.proto.DliFieldTypeProtos;

import eu.dnetlib.data.proto.DliProtos;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.dli.proto.DNGFDLISummaryConverter;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.junit.Before;

import org.junit.Test;

import static eu.dnetlib.data.graph.utils.RelDescriptor.QUALIFIER_SEPARATOR;

/**

 * Created by sandro on 2/13/17.

 */

public class DLIConvertertTest extends AbstractTransformerTest {

	private static final Log log = LogFactory.getLog(DLIConvertertTest.class);

	private Ontologies ontologies;

    @Before

    public void setUp() throws Exception {

        factory = new XsltRowTransformerFactory();

        ontologies = OntologyLoader.loadOntologiesFromCp();

    }

    @Test

    public void testLinkPangaeaDLI() throws Exception {

        final List<Row> rows = Lists.newArrayList();

        rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf.xml")));

        rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf2.xml")));

        rows.addAll(asRows(loadFromTransformationProfile("pmfdli2hbase.xml"), load("record_dli_pmf.xml")));

        printAll(mapAll(buildTable(rows)));

    }

	private void printAll(final Map<String, DNGFDLISummaryConverter> converters) {

		converters.forEach((id, converter) -> log.info(JsonFormat.printToString(converter.convert())));

	}

	private Map<String, DNGFDLISummaryConverter> mapAll(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {

		final Map<String, DNGFDLISummaryConverter> builders = Maps.newHashMap();

		for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {

			map(builders, e.getKey(), e.getValue());

	    }

	    return builders;

    }

	private void map(final Map<String, DNGFDLISummaryConverter> converters, final String rowKey, final Map<String, Map<String, byte[]>> row) throws Exception {

		final Type type = DNGFRowKeyDecoder.decode(rowKey).getType();

		final Map<String, byte[]> familyMap = row.get("metadata");

		if (familyMap == null) return;

		final byte[] bodyB = familyMap.get(type.toString());

		if (bodyB != null) {

			ensureBuilder(converters, rowKey);

			final DNGF mainEntity = DNGFDecoder.decode(

					familyMap.get(type.toString()),

					DliFieldTypeProtos.completionStatus, DliProtos.completionStatus, DliProtos.resolvedfrom, DliProtos.typedIdentifier).getDNGF();

			converters.get(rowKey).setMainEntity(mainEntity);

			final Map<String, byte[]> rels = row.get("rels");

			rels.forEach((q, v) -> {

				final DNGF.Builder rel = DNGF.newBuilder(HBaseTableDAO.parseProto(v));

				rel.getRelBuilder().setCachedTarget(mainEntity.getEntity());

				final String source = rel.getRel().getSource();

				final Type sourceType = rel.getRel().getSourceType();

				rel.getRelBuilder().setSource(rel.getRel().getTarget());

				rel.getRelBuilder().setSourceType(rel.getRel().getTargetType());

				rel.getRelBuilder().setTarget(source);

				rel.getRelBuilder().setTargetType(sourceType);

				final Ontology o = ontologies.get(rel.getRel().getRelType().getSchemeid());

				final String inverse = o.inverseOf(rel.getRel().getRelType().getClassid());

				rel.getRelBuilder().getRelTypeBuilder().setClassid(inverse).setClassname(inverse);

				final String targetId = StringUtils.substringAfter(q, QUALIFIER_SEPARATOR);

				ensureBuilder(converters, targetId);

				converters.get(targetId).addRelation(rel.build());

			});

		}

	}

	private void ensureBuilder(final Map<String, DNGFDLISummaryConverter> builders, final String rowKey) {

		if (!builders.containsKey(rowKey)) {

			builders.put(rowKey, new DNGFDLISummaryConverter());

		}

	}

}