useSourceRefAsSec for Dwc-A in
[cdmlib.git] / cdmlib-io / src / main / java / eu / etaxonomy / cdm / io / dwca / in / DwcTaxonCsv2CdmTaxonConverter.java
index f89c61142ef1e6bbbf7acfe20ae6728afe3d4c33..9e4a9797f3d65e3f54764c87d0efe433e79c2664 100644 (file)
 package eu.etaxonomy.cdm.io.dwca.in;\r
 \r
 import java.util.ArrayList;\r
+import java.util.HashSet;\r
 import java.util.List;\r
 import java.util.Map;\r
+import java.util.Set;\r
 \r
 import org.apache.commons.lang.StringUtils;\r
 import org.apache.log4j.Logger;\r
 \r
+import com.ibm.lsid.MalformedLSIDException;\r
+\r
 import eu.etaxonomy.cdm.common.CdmUtils;\r
 import eu.etaxonomy.cdm.io.dwca.TermUri;\r
 import eu.etaxonomy.cdm.model.common.CdmBase;\r
 import eu.etaxonomy.cdm.model.common.IdentifiableSource;\r
+import eu.etaxonomy.cdm.model.common.LSID;\r
+import eu.etaxonomy.cdm.model.name.BotanicalName;\r
 import eu.etaxonomy.cdm.model.name.NomenclaturalCode;\r
+import eu.etaxonomy.cdm.model.name.NonViralName;\r
 import eu.etaxonomy.cdm.model.name.Rank;\r
 import eu.etaxonomy.cdm.model.name.TaxonNameBase;\r
+import eu.etaxonomy.cdm.model.name.ZoologicalName;\r
 import eu.etaxonomy.cdm.model.reference.Reference;\r
 import eu.etaxonomy.cdm.model.reference.ReferenceFactory;\r
 import eu.etaxonomy.cdm.model.taxon.Classification;\r
 import eu.etaxonomy.cdm.model.taxon.Synonym;\r
 import eu.etaxonomy.cdm.model.taxon.Taxon;\r
 import eu.etaxonomy.cdm.model.taxon.TaxonBase;\r
+import eu.etaxonomy.cdm.strategy.exceptions.StringNotParsableException;\r
 import eu.etaxonomy.cdm.strategy.exceptions.UnknownCdmTypeException;\r
-import eu.etaxonomy.cdm.strategy.parser.INonViralNameParser;\r
 import eu.etaxonomy.cdm.strategy.parser.NonViralNameParserImpl;\r
 \r
 /**\r
@@ -38,28 +46,29 @@ import eu.etaxonomy.cdm.strategy.parser.NonViralNameParserImpl;
  * @date 22.11.2011\r
  *\r
  */\r
-public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState> implements IConverter<CsvStreamItem, IReader<CdmBase>, String>{\r
+public class DwcTaxonCsv2CdmTaxonConverter extends PartitionableConverterBase<DwcaImportState> implements IPartitionableConverter<CsvStreamItem, IReader<CdmBase>, String>{\r
        @SuppressWarnings("unused")\r
        private static Logger logger = Logger.getLogger(DwcTaxonCsv2CdmTaxonConverter.class);\r
 \r
        private static final String ID = "id";\r
-       // key for for case that no dataset information is supplied\r
+       // temporary key for the case that no dataset information is supplied, TODO use something better\r
        public static final String NO_DATASET = "no_dataset_jli773oebhjklw";\r
 \r
+       private NonViralNameParserImpl parser = NonViralNameParserImpl.NewInstance();\r
        \r
        /**\r
         * @param state\r
         */\r
        public DwcTaxonCsv2CdmTaxonConverter(DwcaImportState state) {\r
-               super();\r
-               this.state = state;\r
+               super(state);\r
        }\r
 \r
 \r
        public IReader<MappedCdmBase> map(CsvStreamItem csvTaxonRecord){\r
                List<MappedCdmBase> resultList = new ArrayList<MappedCdmBase>(); \r
                \r
-               Reference<?> sourceReference = null;\r
+               //TODO what if not transactional? \r
+               Reference<?> sourceReference = state.getTransactionalSourceReference();\r
                String sourceReferenceDetail = null;\r
                \r
                //taxon\r
@@ -67,7 +76,7 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
                MappedCdmBase  mcb = new MappedCdmBase(csvTaxonRecord.term, csvTaxonRecord.get(ID), taxonBase);\r
                resultList.add(mcb);\r
                \r
-               //source\r
+               //original source\r
                String id = csvTaxonRecord.get(ID);\r
                IdentifiableSource source = taxonBase.addSource(id, "Taxon", sourceReference, sourceReferenceDetail);\r
                MappedCdmBase mappedSource = new MappedCdmBase(csvTaxonRecord.get(ID), source);\r
@@ -78,24 +87,29 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
                NomenclaturalCode nomCode = getNomCode(csvTaxonRecord);\r
                Rank rank = getRank(csvTaxonRecord, nomCode);\r
 \r
-               //name\r
-               TaxonNameBase<?,?> name = getScientificName(csvTaxonRecord, nomCode, rank);\r
+               //name && name published in\r
+               TaxonNameBase<?,?> name = getScientificName(csvTaxonRecord, nomCode, rank, resultList, sourceReference);\r
                taxonBase.setName(name);\r
                \r
-               //sec\r
-               Reference<?> sec = getNameAccordingTo(csvTaxonRecord);\r
-               taxonBase.setSec(sec);\r
+               //nameAccordingTo\r
+               MappedCdmBase<Reference> sec = getNameAccordingTo(csvTaxonRecord, resultList);\r
+               if (sec == null && state.getConfig().isUseSourceReferenceAsSec()){\r
+                       sec = new MappedCdmBase<Reference>(state.getTransactionalSourceReference());\r
+               }\r
+               if (sec != null){\r
+                       taxonBase.setSec(sec.getCdmBase());\r
+               }\r
 \r
                //classification\r
-               handleDataset(csvTaxonRecord, resultList, sourceReference, sourceReferenceDetail);\r
+               handleDataset(csvTaxonRecord, taxonBase, resultList, sourceReference, sourceReferenceDetail);\r
                \r
+               //NON core\r
+           //term="http://purl.org/dc/terms/identifier"\r
+               //currently only LSIDs\r
+               handleIdentifier(csvTaxonRecord, taxonBase); \r
+\r
                \r
-//                 <field index="0" term="http://rs.tdwg.org/dwc/terms/taxonID"/>\r
                \r
-//                 <!-- LSID -->\r
-//                 <field index="1" term="http://purl.org/dc/terms/identifier"/>\r
-\r
-\r
                //                  <!-- Top level group; listed as kingdom but may be interpreted as domain or superkingdom\r
 //                      The following eight groups are recognized: Animalia, Archaea, Bacteria, Chromista, \r
 //                      Fungi, Plantae, Protozoa, Viruses -->\r
@@ -127,6 +141,7 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
 //                 <!-- Authorship -->\r
 \r
 //                 <field index='19' term='http://rs.tdwg.org/dwc/terms/scientificNameAuthorship'/>\r
+//             ==> see scientific name\r
 //                 \r
 //             <!-- Acceptance status published in -->\r
 //                 <field index='20' term='http://purl.org/dc/terms/source'/>\r
@@ -144,37 +159,89 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
        }\r
 \r
 \r
-       private void handleDataset(CsvStreamItem csvTaxonRecord, List<MappedCdmBase> resultList, Reference<?> sourceReference, String sourceReferecenDetail) {\r
-               String datasetId = CdmUtils.Nz(csvTaxonRecord.get(TermUri.DWC_DATASET_ID)).trim();\r
-               String datasetName = CdmUtils.Nz(csvTaxonRecord.get(TermUri.DWC_DATASET_NAME)).trim();\r
-               if (CdmUtils.areBlank(datasetId, datasetName) ){\r
-                       datasetId = NO_DATASET;\r
+       \r
+       //TODO handle non LSIDs\r
+       //TODO handle LSIDs for names\r
+       private void handleIdentifier(CsvStreamItem csvTaxonRecord, TaxonBase<?> taxonBase) {\r
+               String identifier = csvTaxonRecord.get(TermUri.DC_IDENTIFIER);\r
+               if (StringUtils.isNotBlank(identifier)){\r
+                       if (identifier.trim().startsWith("urn:lsid")){\r
+                               try {\r
+                                       LSID lsid = new LSID(identifier);\r
+                                       taxonBase.setLsid(lsid);\r
+                               } catch (MalformedLSIDException e) {\r
+                                       String message = "LSID is malformed and can't be handled as LSID: %s";\r
+                                       message = String.format(message, identifier);\r
+                                       fireWarningEvent(message, csvTaxonRecord, 4);\r
+                               } \r
+                       }else{\r
+                               String message = "Identifier type not supported: %s";\r
+                               message = String.format(message, identifier);\r
+                               fireWarningEvent(message, csvTaxonRecord, 4);\r
+                       }\r
                }\r
                \r
-               //check id\r
-               boolean classificationExists = state.exists(TermUri.DWC_DATASET_ID.toString() , datasetId, Classification.class);\r
-               \r
-               //check name\r
-               if (!classificationExists){\r
-                       classificationExists = state.exists(TermUri.DWC_DATASET_ID.toString() , datasetName, Classification.class);\r
-               }\r
+       }\r
+\r
+\r
+       private void handleDataset(CsvStreamItem item, TaxonBase<?> taxonBase, List<MappedCdmBase> resultList, Reference<?> sourceReference, String sourceReferecenDetail) {\r
+               if (config.isDatasetsAsClassifications()){\r
+                       String datasetId = CdmUtils.Nz(item.get(TermUri.DWC_DATASET_ID)).trim();\r
+                       String datasetName = CdmUtils.Nz(item.get(TermUri.DWC_DATASET_NAME)).trim();\r
+                       if (CdmUtils.areBlank(datasetId, datasetName) ){\r
+                               datasetId = NO_DATASET;\r
+                       }\r
+                       \r
+                       //check id\r
+                       boolean classificationExists = state.exists(TermUri.DWC_DATASET_ID.toString() , datasetId, Classification.class);\r
+                       \r
+                       //check name\r
+                       if (!classificationExists){\r
+                               classificationExists = state.exists(TermUri.DWC_DATASET_NAME.toString() , datasetName, Classification.class);\r
+                       }\r
+                       \r
+                       //if not exists, create new\r
+                       if (! classificationExists){\r
+                               String classificationName = StringUtils.isBlank(datasetName)? datasetId : datasetName;\r
+                               if (classificationName.equals(NO_DATASET)){\r
+                                       classificationName = "Classification (no name)";  //TODO define by config or zipfile or metadata\r
+                               }\r
+                               \r
+                               String classificationId = StringUtils.isBlank(datasetId)? datasetName : datasetId;\r
+                               Classification classification = Classification.NewInstance(classificationName);\r
+                               //source\r
+                               IdentifiableSource source = classification.addSource(classificationId, "Dataset", sourceReference, sourceReferecenDetail);\r
+                               //add to result\r
+                               resultList.add(new MappedCdmBase(TermUri.DWC_DATASET_ID, datasetId, classification));\r
+                               resultList.add(new MappedCdmBase(TermUri.DWC_DATASET_NAME, datasetName, classification));\r
+                               resultList.add(new MappedCdmBase(source));\r
+                               //TODO this is not so nice but currently necessary as classifications are requested in the same partition\r
+                               state.putMapping(TermUri.DWC_DATASET_ID.toString(), classificationId, classification);\r
+                               state.putMapping(TermUri.DWC_DATASET_NAME.toString(), classificationName, classification);\r
+                       }\r
+               }else if (config.isDatasetsAsSecundumReference()){\r
+                       //dataset as secundum reference\r
+                       TermUri idTerm = TermUri.DWC_DATASET_ID;\r
+                       TermUri strTerm = TermUri.DWC_DATASET_NAME;\r
+                       MappedCdmBase<Reference> mappedCitation = getReference(item, resultList, idTerm, strTerm, true);\r
+                       Reference<?> sec = mappedCitation.getCdmBase();\r
+                       taxonBase.setSec(sec);\r
                \r
-               //if not exists, create new\r
-               if (! classificationExists){\r
-                       String classificationName = StringUtils.isBlank(datasetName)? datasetId : datasetName;\r
-                       String classificationId = StringUtils.isBlank(datasetId)? datasetName : datasetId;\r
-                       Classification classification = Classification.NewInstance(classificationName);\r
-                       //source\r
-                       IdentifiableSource source = classification.addSource(classificationId, "Dataset", sourceReference, sourceReferecenDetail);\r
-                       //add to result\r
-                       resultList.add(new MappedCdmBase(TermUri.DWC_DATASET_ID, datasetId, classification));\r
-                       resultList.add(new MappedCdmBase(TermUri.DWC_DATASET_NAME, datasetName, classification));\r
-                       resultList.add(new MappedCdmBase(source));\r
+               }else if (config.isDatasetsAsOriginalSource()){\r
+                       //dataset as original source\r
+                       TermUri idTerm = TermUri.DWC_DATASET_ID;\r
+                       TermUri strTerm = TermUri.DWC_DATASET_NAME;\r
+                       MappedCdmBase<Reference> mappedCitation = getReference(item, resultList, idTerm, strTerm, true);\r
+//                     resultList.add(mappedCitation);\r
+                       taxonBase.addSource(null, null, mappedCitation.getCdmBase(), null);\r
+               }else{\r
+                       String message = "DatasetUse type not yet implemented. Can't import dataset information.";\r
+                       fireWarningEvent(message, item, 4);\r
                }\r
                \r
                //remove to later check if all attributes were used\r
-               csvTaxonRecord.remove(TermUri.DWC_DATASET_ID);\r
-               csvTaxonRecord.remove(TermUri.DWC_DATASET_NAME);\r
+               item.remove(TermUri.DWC_DATASET_ID);\r
+               item.remove(TermUri.DWC_DATASET_NAME);\r
                \r
        }\r
 \r
@@ -185,52 +252,186 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
                return id;\r
        }\r
 \r
-       private Reference<?> getNameAccordingTo(CsvStreamItem csvTaxonRecord) {\r
-               String strSec = csvTaxonRecord.get(TermUri.DWC_NAME_ACCORDING_TO);\r
-               if (strSec != null){\r
-                       Reference<?> sec = ReferenceFactory.newGeneric();\r
-                       sec.setTitleCache(strSec, true);\r
-                       return sec;\r
+       private MappedCdmBase<Reference> getNameAccordingTo(CsvStreamItem item, List<MappedCdmBase> resultList) {\r
+               if (config.isDatasetsAsSecundumReference()){\r
+                       //TODO store nameAccordingTo info some where else or let the user define where to store it.\r
+                       return null;\r
+               }else{\r
+                       TermUri idTerm = TermUri.DWC_NAME_ACCORDING_TO_ID;\r
+                       TermUri strTerm = TermUri.DWC_NAME_ACCORDING_TO;\r
+                       MappedCdmBase<Reference> secRef = getReference(item, resultList, idTerm, strTerm, false);\r
+                       return secRef;\r
                }\r
-               return null;\r
        }\r
 \r
-\r
        private NomenclaturalCode getNomCode(CsvStreamItem item) {\r
                String strNomCode = getValue(item, TermUri.DWC_NOMENCLATURAL_CODE);\r
+               NomenclaturalCode nomCode = null;\r
+               // by Nomcenclatural Code\r
                if (strNomCode != null){\r
-                       NomenclaturalCode nomCode = NomenclaturalCode.fromString(strNomCode);\r
+                       nomCode = NomenclaturalCode.fromString(strNomCode);\r
                        if (nomCode == null){\r
                                String message = "NomCode '%s' not recognized";\r
                                message = String.format(message, strNomCode);\r
                                fireWarningEvent(message, item, 4);\r
+                       }else{\r
+                               return nomCode;\r
+                       }\r
+               }\r
+               // by Kingdom\r
+               String strKingdom = getValue(item, TermUri.DWC_KINGDOM);\r
+               if (strKingdom != null){\r
+                       if (strKingdom.equalsIgnoreCase("Plantae")){\r
+                               nomCode = NomenclaturalCode.ICBN;\r
+                       }else if (strKingdom.equalsIgnoreCase("Fungi")){\r
+                               nomCode = NomenclaturalCode.ICBN;\r
+                       }else if (strKingdom.equalsIgnoreCase("Animalia")){\r
+                               nomCode = NomenclaturalCode.ICZN;\r
+                       }else if (strKingdom.equalsIgnoreCase("Protozoa")){\r
+                               nomCode = NomenclaturalCode.ICZN;\r
+                       }\r
+               }\r
+               \r
+               //TODO further kingdoms\r
+               if (nomCode == null){\r
+                       //TODO warning\r
+                       if (config.getNomenclaturalCode() != null){\r
+                               nomCode = config.getNomenclaturalCode();\r
                        }\r
-                       return nomCode;\r
                }\r
-               return null;\r
+               return nomCode;\r
        }\r
 \r
 \r
-       private TaxonNameBase<?,?> getScientificName(CsvStreamItem item, NomenclaturalCode nomCode, Rank rank) {\r
+       private TaxonNameBase<?,?> getScientificName(CsvStreamItem item, NomenclaturalCode nomCode, Rank rank, List<MappedCdmBase> resultList, Reference sourceReference) {\r
+               TaxonNameBase<?,?> name = null;\r
                String strScientificName = getValue(item, TermUri.DWC_SCIENTIFIC_NAME);\r
+               //Name\r
                if (strScientificName != null){\r
-                       INonViralNameParser<?> parser = NonViralNameParserImpl.NewInstance();\r
-                       TaxonNameBase<?,?> name = parser.parseFullName(strScientificName, nomCode, rank);\r
-                       if (rank != null && name != null && name.getRank() != null && \r
-                                       ! rank.equals(name.getRank())){\r
-                               String message = "Parsed rank %s differs from given rank %s";\r
-                               message = String.format(message, name.getRank().getTitleCache(), rank.getTitleCache());\r
-                               fireWarningEvent(message, item, 4);\r
+                       name = parser.parseFullName(strScientificName, nomCode, rank);\r
+                       if ( rank != null && name != null && name.getRank() != null &&  ! rank.equals(name.getRank())){\r
+                               if (config.isValidateRankConsistency()){\r
+                                       String message = "Parsed rank %s (%s) differs from rank %s given by fields 'taxonRank' or 'verbatimTaxonRank'";\r
+                                       message = String.format(message, name.getRank().getTitleCache(), strScientificName, rank.getTitleCache());\r
+                                       fireWarningEvent(message, item, 4);\r
+                               }\r
                        }\r
-                       return name;\r
+                       checkAuthorship(name, item);\r
+                       resultList.add(new MappedCdmBase(TermUri.DWC_SCIENTIFIC_NAME, strScientificName, name));\r
                }\r
+               //By ID\r
                String strScientificNameId = getValue(item, TermUri.DWC_SCIENTIFIC_NAME_ID);\r
                if (strScientificNameId != null){\r
-                       String message = "ScientificNameId not yet implemented: '%s'";\r
-                       message = String.format(message, strScientificNameId);\r
-                       fireWarningEvent(message, item, 4);\r
+                       if (config.isScientificNameIdAsOriginalSourceId()){\r
+                               if (name != null){\r
+                                       IdentifiableSource source = IdentifiableSource.NewInstance(strScientificNameId, TermUri.DWC_SCIENTIFIC_NAME_ID.toString(), sourceReference, null);\r
+                                       name.addSource(source);\r
+                               }\r
+                       }else{\r
+                               String message = "ScientificNameId not yet implemented: '%s'";\r
+                               message = String.format(message, strScientificNameId);\r
+                               fireWarningEvent(message, item, 4);\r
+                       }\r
+               }\r
+               \r
+               //namePublishedIn\r
+               TermUri idTerm = TermUri.DWC_NAME_PUBLISHED_IN_ID;\r
+               TermUri strTerm = TermUri.DWC_NAME_PUBLISHED_IN;\r
+               MappedCdmBase<Reference> nomRef = getReference(item, resultList, idTerm, strTerm, false);\r
+               \r
+               if (name != null){\r
+                       if (nomRef != null){\r
+                               name.setNomenclaturalReference(nomRef.getCdmBase());  //check if name already has a nomRef, shouldn't be the case usually\r
+                       }\r
+               }else{\r
+                       if (nomRef != null){\r
+                               String message = "NamePublishedIn information available but no name exists";\r
+                               fireWarningEvent(message, item, 4);\r
+                       }\r
                }\r
-               return null;\r
+               return name;\r
+       }\r
+\r
+\r
+       /**\r
+        * General method to handle references used for multiple attributes.\r
+        * @param item\r
+        * @param resultList\r
+        * @param idTerm\r
+        * @param strTerm\r
+        * @param idIsInternal\r
+        * @return\r
+        */\r
+       private MappedCdmBase<Reference> getReference(CsvStreamItem item, List<MappedCdmBase> resultList, TermUri idTerm, TermUri strTerm, boolean idIsInternal) {\r
+               Reference<?> newRef = null;\r
+               Reference<?> sourceCitation = null;\r
+               \r
+               MappedCdmBase<Reference> result = null;\r
+               if (exists(idTerm, item) || exists(strTerm, item)){\r
+                       String refId = CdmUtils.Nz(item.get(idTerm)).trim();\r
+                       String refStr = CdmUtils.Nz(item.get(strTerm)).trim();\r
+                       if (StringUtils.isNotBlank(refId)){\r
+                               List<Reference> references = state.get(idTerm.toString(), refId, Reference.class);\r
+                               if (references.size() == 0){\r
+                                       if (! idIsInternal){\r
+                                               //references should already exist in store if not linking to external links like URLs\r
+                                               String message = "External namePublishedInIDs are not yet supported";\r
+                                               fireWarningEvent(message, item, 4);\r
+                                       }else{\r
+                                               newRef = ReferenceFactory.newGeneric();  //TODO handle other types if possible\r
+                                               newRef.addSource(refId, idTerm.toString(), sourceCitation, null);\r
+                                               MappedCdmBase<Reference> idResult = new MappedCdmBase<Reference>(idTerm, refId, newRef);\r
+                                               resultList.add(idResult);\r
+                                       }\r
+                               }else{\r
+                                       //TODO handle list.size > 1 , do we need a list here ?\r
+                                       result = new MappedCdmBase<Reference>(idTerm, refId , references.get(0));\r
+                               }\r
+                       }\r
+                       if (result == null){\r
+                               List<Reference> nomRefs = state.get(strTerm.toString(), refStr, Reference.class);\r
+                               if (nomRefs.size() > 0){\r
+                                       //TODO handle list.size > 1 , do we need a list here ?\r
+                                       result = new MappedCdmBase<Reference>(strTerm, refStr , nomRefs.get(0));\r
+                               }else{\r
+                                       // new Reference\r
+                                       if (newRef == null){\r
+                                               newRef = ReferenceFactory.newGeneric();  //TODO handle other types if possible\r
+                                       }\r
+                                       newRef.setTitleCache(refStr, true);\r
+                                       //TODO distinguish available year, authorship, etc. if\r
+                                       result = new MappedCdmBase<Reference>(strTerm, refStr, newRef);\r
+                                       resultList.add(result);\r
+                               }\r
+                       }\r
+               }\r
+               return result;\r
+       }\r
+\r
+\r
+       //TODO we may configure in configuration that scientific name never includes Authorship\r
+       private void checkAuthorship(TaxonNameBase nameBase, CsvStreamItem item) {\r
+               if (!nameBase.isInstanceOf(NonViralName.class)){\r
+                       return;\r
+               }\r
+               NonViralName<?> nvName = CdmBase.deproxy(nameBase, NonViralName.class); \r
+               String strAuthors = getValue(item, TermUri.DWC_SCIENTIFIC_NAME_AUTHORS);\r
+               \r
+               if (! nvName.isProtectedTitleCache()){\r
+                       if (StringUtils.isBlank(nvName.getAuthorshipCache())){\r
+                               if (nvName.isInstanceOf(BotanicalName.class) || nvName.isInstanceOf(ZoologicalName.class)){\r
+                                       try {\r
+                                               parser.parseAuthors(nvName, strAuthors);\r
+                                       } catch (StringNotParsableException e) {\r
+                                               nvName.setAuthorshipCache(strAuthors);\r
+                                       }               \r
+                               }else{\r
+                                       nvName.setAuthorshipCache(strAuthors);\r
+                               }\r
+                               //TODO throw warning (scientific name should always include authorship) by DwC definition\r
+                       }\r
+               }\r
+               \r
        }\r
 \r
 \r
@@ -272,6 +473,11 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
        }\r
 \r
 \r
+       /**\r
+        * Creates an empty taxon object with a given status.\r
+        * @param item\r
+        * @return\r
+        */\r
        private TaxonBase<?> getTaxonBase(CsvStreamItem item) {\r
                TaxonNameBase<?,?> name = null;\r
                Reference<?> sec = null;\r
@@ -313,11 +519,75 @@ public class DwcTaxonCsv2CdmTaxonConverter extends ConverterBase<DwcaImportState
                }\r
                        \r
                return result;\r
+\r
        }\r
        \r
+// ********************** PARTITIONABLE ****************************************/\r
+\r
+\r
+       @Override\r
+       protected void makeForeignKeysForItem(CsvStreamItem item, Map<String, Set<String>> fkMap) {\r
+               String value;\r
+               String key;\r
+               \r
+               //namePublishedIn\r
+               if ( hasValue(value = item.get(key = TermUri.DWC_NAME_PUBLISHED_IN_ID.toString()))){\r
+                       Set<String> keySet = getKeySet(key, fkMap);\r
+                       keySet.add(value);\r
+               }\r
+               if (config.isDeduplicateNamePublishedIn()){\r
+                       if ( hasValue(value = item.get(key = TermUri.DWC_NAME_PUBLISHED_IN.toString()))){\r
+                               Set<String> keySet = getKeySet(key, fkMap);\r
+                               keySet.add(value);\r
+                       }\r
+               }\r
+               \r
+               //nameAccordingTo\r
+               if (! config.isDatasetsAsSecundumReference()){\r
+                       if ( hasValue(value = item.get(key = TermUri.DWC_NAME_ACCORDING_TO_ID.toString()))){\r
+                               Set<String> keySet = getKeySet(key, fkMap);\r
+                               keySet.add(value);\r
+                       }\r
+                       if ( hasValue(value = item.get(key = TermUri.DWC_NAME_ACCORDING_TO.toString()))){\r
+                               Set<String> keySet = getKeySet(key, fkMap);\r
+                               keySet.add(value);\r
+                       }\r
+               }\r
+               \r
+               //dataset\r
+               if ( hasValue(value = item.get(key = TermUri.DWC_DATASET_ID.toString()))){\r
+                       Set<String> keySet = getKeySet(key, fkMap);\r
+                       keySet.add(value);\r
+               }\r
+               if ( hasValue(value = item.get(key = TermUri.DWC_DATASET_NAME.toString()))){\r
+                       Set<String> keySet = getKeySet(key, fkMap);\r
+                       keySet.add(value);\r
+               }\r
+               \r
+       }\r
+       \r
+       \r
+       @Override\r
+       public Set<String> requiredSourceNamespaces() {\r
+               Set<String> result = new HashSet<String>();\r
+               result.add(TermUri.DWC_NAME_PUBLISHED_IN_ID.toString());\r
+               result.add(TermUri.DWC_NAME_PUBLISHED_IN.toString());\r
+               if (!config.isDatasetsAsSecundumReference()){\r
+                       result.add(TermUri.DWC_NAME_ACCORDING_TO_ID.toString());\r
+                       result.add(TermUri.DWC_NAME_ACCORDING_TO.toString());\r
+               }\r
+               result.add(TermUri.DWC_DATASET_ID.toString());\r
+               result.add(TermUri.DWC_DATASET_NAME.toString());\r
+               return result;\r
+       }\r
+       \r
+//** ***************************** TO STRING *********************************************/\r
+       \r
        @Override\r
        public String toString(){\r
                return this.getClass().getName();\r
        }\r
 \r
+\r
+       \r
 }\r