Changeset 625

Timestamp:

Mar 11, 2008, 2:36:19 PM (15 years ago)

Author:

Nicklas Nordborg

Message:

References #102: Intensity calculator for Illumina SNP data

Added error handling. Improved performance by preloading reporters.

Location:

plugins/base2/net.sf.basedb.illumina/trunk/src/net/sf/basedb/illumina/plugins

Files:

• ReporterProxy.java (added)
• SnpIntensityCalculator.java (modified) (18 diffs)

plugins/base2/net.sf.basedb.illumina/trunk/src/net/sf/basedb/illumina/plugins/SnpIntensityCalculator.java

@@ -25 +25 @@
 
 import java.io.IOException;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
@@ -38 +42 @@
 import net.sf.basedb.core.Config;
 import net.sf.basedb.core.DbControl;
+import net.sf.basedb.core.DynamicReporterQuery;
+import net.sf.basedb.core.DynamicResultIterator;
 import net.sf.basedb.core.Experiment;
 import net.sf.basedb.core.File;
@@ -45 +51 @@
 import net.sf.basedb.core.InvalidUseOfNullException;
 import net.sf.basedb.core.Item;
+import net.sf.basedb.core.ItemNotFoundException;
 import net.sf.basedb.core.ItemParameterType;
 import net.sf.basedb.core.ItemQuery;
@@ -54 +61 @@
 import net.sf.basedb.core.RawBioAssay;
 import net.sf.basedb.core.RawDataType;
-import net.sf.basedb.core.ReporterBatcher;
+import net.sf.basedb.core.Reporter;
 import net.sf.basedb.core.RequestInformation;
+import net.sf.basedb.core.SimpleAbsoluteProgressReporter;
 import net.sf.basedb.core.SpotBatcher;
 import net.sf.basedb.core.StringParameterType;
+import net.sf.basedb.core.StringUtil;
 import net.sf.basedb.core.Transformation;
 import net.sf.basedb.core.data.ReporterData;
@@ -68 +77 @@
 import net.sf.basedb.core.plugin.Request;
 import net.sf.basedb.core.plugin.Response;
+import net.sf.basedb.core.query.Dynamic;
 import net.sf.basedb.core.query.Hql;
 import net.sf.basedb.core.query.Orders;
+import net.sf.basedb.core.query.SqlResult;
 import net.sf.basedb.core.signal.SignalHandler;
 import net.sf.basedb.core.signal.SignalTarget;
@@ -75 +86 @@
 import net.sf.basedb.illumina.Illumina;
 import net.sf.basedb.plugins.util.Parameters;
-import net.sf.basedb.util.ChainedProgressReporter;
 import net.sf.basedb.util.IntensityCalculatorUtil;
 import net.sf.basedb.util.NumberFormatUtil;
@@ -104 +114 @@
       "as follows:\n" +
       " * Channel 1 = GType (AA = 1.0, AB = 0.0, BB = -1.0, NC = null\n" +
-      " * Channel 2 = B Allele Freq\n" +
-      " * Channel 3 = Log R Ratio",
+      " * Channel 2 = Log R Ratio\n" +
+      " * Channel 3 = B Allele Freq",
       Illumina.VERSION,
       Illumina.COPYRIGHT,
@@ -123 +133 @@
   
   
+  private static final PluginParameter<String> missingReporterErrorParameter = new PluginParameter<String>(
+    "missingReporterError",
+    "Reporter not found",
+    "How to handle errors that are caused by a reporter not beeing present in the datbase. If not specified the " +
+      "default error handling is used.\n\n"+
+      "null = Import the data but set the reporter to null\n"+
+      "skip = Skip the current data line and continue\n"+
+      "fail = Stop with an error message",
+      new StringParameterType(255, null, false, 1, 0, 0, 
+          Arrays.asList( new String[] { "null", "skip", "fail"} ))
+    );
+
   private RequestInformation configureJob;
   private ThreadSignalHandler signalHandler;
@@ -175 +197 @@
       // rawBioAssays parameter
       List<RawBioAssay> sources = (List<RawBioAssay>)job.getValues("rawBioAssays");
+      long totalProgressTicks = 0;
       if (sources == null || sources.size() == 0)
       {
@@ -182 +205 @@
       {
         dc.reattachItem(rba);
+        totalProgressTicks += rba.getNumFileSpots();
       }
 
@@ -197 +221 @@
       dc.saveItem(rootBas);
 
-      ChainedProgressReporter chainedProgress = progress == null ? 
-          null : new ChainedProgressReporter(progress);
-      float percentPerStep = 100.0f / (1 + sources.size());
-      int lowLimit = 0;
-      int highLimit = (int)percentPerStep;
-
-      // Create bioassays for each source
+      ArrayDesign design = sources.get(0).getArrayDesign();
+      totalProgressTicks += 3 * design.getNumFileFeatures();
+      SimpleAbsoluteProgressReporter simpleProgress = progress == null ? 
+        null : new SimpleAbsoluteProgressReporter(progress, totalProgressTicks);
+      
+      // 1. Preload all reporters from the SNP manifest file
+      Map<String, ReporterProxy> preloaded = null;
+      preloaded = preloadReporters(dc, design, simpleProgress);
+      
+      // 2. Load reporter ID:s from the database
+      mapReporterIds(dc, preloaded, simpleProgress);
+      
+      // 3. Insert position data into the database
+      PositionBatcher posBatcher = rootBas.getPositionBatcher();
+      Set<Integer> skipped = insertPosData(dc, posBatcher, preloaded, simpleProgress);
+      posBatcher.close();
+      
+      // 4. Create bioassays for each raw bioassay source
       SpotBatcher spotBatcher = rootBas.getSpotBatcher();
+      int numInserted = 0;
       for (RawBioAssay rba : sources)
       {
-        if (chainedProgress != null)
-        {
-          chainedProgress.setRange(lowLimit, highLimit);
-          lowLimit = highLimit;
-          highLimit += percentPerStep;
-        }
-        insertSpotData(dc, rba, rootBas, spotBatcher, chainedProgress);
+        numInserted += insertSpotData(dc, rba, rootBas, spotBatcher, skipped, simpleProgress);
       }
       spotBatcher.close();      
-      
-      // Get the manifest file from the array design and 
-      // batch position, reporter and addressID
-      if (chainedProgress != null)
-      {
-        chainedProgress.setRange(lowLimit, highLimit);
-        lowLimit = highLimit;
-        highLimit += percentPerStep;
-      }
-      PositionBatcher posBatcher = rootBas.getPositionBatcher();
-      ReporterBatcher reporters = ReporterBatcher.getNew(dc);
-      ArrayDesign design = sources.get(0).getArrayDesign();
-      insertPosData(dc, design, posBatcher, reporters, chainedProgress);
-      posBatcher.close();
-      reporters.close();
-  
+      if (progress != null) progress.display(100, "Committing transactio...");
       dc.commit();
       if (progress != null) progress.display(100, "Done");
+      String message = numInserted + " spots inserted";
+      if (!skipped.isEmpty()) message += "; " + skipped.size() + " positions skipped (missing reporter)";
       response.setDone("Done");
     }
@@ -342 +359 @@
         storeValue(job, request, ri.getParameter(Parameters.CHARSET_PARAMETER));
         storeValue(job, request, ri.getParameter(Parameters.DECIMAL_SEPARATOR_PARAMETER));
+        
+        // Error handling
+        storeValue(job, request, ri.getParameter(Parameters.DEFAULT_ERROR));
+        storeValue(job, request, ri.getParameter(Parameters.NUMBER_FORMAT_ERROR));
+        storeValue(job, request, ri.getParameter(missingReporterErrorParameter.getName()));
 
         Job.ExecutionTime execTime = Job.ExecutionTime.MEDIUM;  
@@ -382 +404 @@
     if (charset == null) charset = Config.getCharset();
     return charset;
+  } 
+  
+  private String getErrorOption(String parameterName)
+  {
+    Object option = null;
+    option = job.getValue(parameterName);
+    if (option == null) option = job.getValue(Parameters.DEFAULT_ERROR);
+    return option == null ? "fail" : option.toString();
   } 
   
@@ -420 +450 @@
         parameters.add(Parameters.charsetParameter(null, null, null));
         parameters.add(Parameters.decimalSeparatorParameter(null, null, null));
+        
+        parameters.add(Parameters.errorSection(null, null));
+        parameters.add(Parameters.defaultError(null, null, "fail"));
+        parameters.add(Parameters.numberFormatError(null, null, null));
+        parameters.add(missingReporterErrorParameter);
       }
       finally
@@ -436 +471 @@
   }
   
-  /**
-    Use the manifest file on the array design to map each adress ID to
-    a reporter.
-  */
-  private void insertPosData(DbControl dc, ArrayDesign design, PositionBatcher posBatcher, 
-    ReporterBatcher reporters, ProgressReporter progress)
-    throws IOException
-  {
-    
+  private File getManifestFile(DbControl dc, ArrayDesign design)
+  {
     File manifest = FileStoreUtil.getDataFile(dc, design, Illumina.SNP_MANIFEST_FILE_ID);
     if (manifest == null)
     {
       throw new BaseException("Can't find manifest file on array design: " + design);
-    }
-    
+    }   
+    return manifest;
+  }
+  
+  private FlatFileParser getManifestParser(File manifest)
+    throws IOException
+  {
     FlatFileParser ffp = new FlatFileParser();
     ffp.setDataSplitterRegexp(Pattern.compile(","));
@@ -461 +494 @@
       throw new BaseException("Can't find SNP data in file: " + manifest);
     }
-    
-    Mapper reporterMapper = ffp.getMapper("\\IlmnID\\");
-    Mapper addressMapper = ffp.getMapper("\\AddressA_ID\\");
-    
-    int numSpots = design.getNumFileFeatures();
-    int currentSpot = 0;
-    int interval = numSpots / 100;
-    float progressFactor = 100.0f / numSpots;
-    while (ffp.hasMoreData())
-    {
-      checkInterrupted();
-      currentSpot++;
-      FlatFileParser.Data data = ffp.nextData();
-      String reporterId = reporterMapper.getValue(data);
-      int addressId = addressMapper.getInt(data);
-      ReporterData reporter = reporters.getByExternalId(reporterId);
-      posBatcher.insert(addressId, reporter);
-      
-      if (currentSpot % interval == 0 && progress != null)
-      {
-        progress.display((int)(progressFactor * currentSpot), 
-          "Mapping positions from array design. " + 
-          currentSpot + " of " + numSpots + " done.");
-      }
-    }
-    posBatcher.flush();
-    
-  }
-  
-  private void insertSpotData(DbControl dc, RawBioAssay rba, BioAssaySet root, 
-    SpotBatcher spotBatcher, ProgressReporter progress)
+    return ffp;
+  }
+  
+
+  private int insertSpotData(DbControl dc, RawBioAssay rba, BioAssaySet root, 
+    SpotBatcher spotBatcher, Set<Integer> skipped, SimpleAbsoluteProgressReporter progress)
     throws IOException
   {
@@ -520 +528 @@
       throw new BaseException("Can't find SNP data in file: " + dataFile);
     }
+
+    String invalidNumber = getErrorOption(Parameters.NUMBER_FORMAT_ERROR);
+    boolean skipLine = "skip".equals(invalidNumber);
+    boolean useNull = "null".equals(invalidNumber);
     
     BioAssay ba = root.newRootBioAssay(Collections.singleton(rba));
@@ -528 +540 @@
     Mapper addressMapper = ffp.getMapper("\\Address\\");
     Mapper gTypeMapper = new GTypeMapper(ffp.getColumnHeaderIndex("GType"));
-    Mapper bAlleleMapper = new Null2NaNMapper(ffp.getMapper("\\B Allele Freq\\", true));
-    Mapper ratioMapper = new Null2NaNMapper(ffp.getMapper("\\Log R Ratio\\", true));
-
+    Mapper bAlleleMapper = new Null2NaNMapper(ffp.getMapper("\\B Allele Freq\\", useNull));
+    Mapper ratioMapper = new Null2NaNMapper(ffp.getMapper("\\Log R Ratio\\", useNull));
+  
     int numSpots = rba.getNumFileSpots();
+    long baseProgress = progress == null ? 0 : progress.getCompleted();
+    int currentSpot = 0;
+    int numInserted = 0;
+    int interval = numSpots / 100;
+    float[] snpdata = new float[3];
+    FlatFileParser.Data dataLine = null;
+    try
+    {
+      while (ffp.hasMoreData())
+      {         
+        checkInterrupted();
+        currentSpot++;
+        dataLine = ffp.nextData();
+        int addressId = addressMapper.getInt(dataLine);
+        if (skipped.contains(addressId)) continue;
+        
+        try
+        {
+          snpdata[0] = gTypeMapper.getFloat(dataLine);
+          snpdata[1] = ratioMapper.getFloat(dataLine);
+          snpdata[2] = bAlleleMapper.getFloat(dataLine);
+        }
+        catch (RuntimeException ex)
+        {
+          if (skipLine) continue;
+          throw ex;
+        }
+        numInserted++;
+        spotBatcher.insert(baCol, addressId, snpdata);
+        
+        if (currentSpot % interval == 0 && progress != null)
+        {
+          progress.displayAbsolute(baseProgress + currentSpot, 
+            "Inserting spot data for raw bioassay '" + rba.getName() + "'. " + 
+            currentSpot + " of " + numSpots + " done.");
+        }
+      }
+      spotBatcher.flush();
+    }
+    catch (Exception ex)
+    {
+      throw new BaseException(ex.getMessage() + 
+          " on line " + (dataLine == null ? "-1" : dataLine.lineNo() + " in file '" +
+          dataFile.getName() + "': " + 
+          StringUtil.trimString(dataLine.line(), 20)), ex);     
+    }
+    return numInserted;
+  }
+  
+  /**
+    Parse the SNP manifest file attached to the array design. Pick the
+    AddressA_ID --> position and IlmnID --> Reporter.externalId and
+    put those in a map which is returned.
+  */
+  private Map<String, ReporterProxy> preloadReporters(DbControl dc, ArrayDesign design, 
+    SimpleAbsoluteProgressReporter progress)
+    throws IOException
+  {
+    int numSpots = design.getNumFileFeatures();
+    long baseProgress = progress == null ? 0 : progress.getCompleted();
+    int currentSpot = 0;
+    int numInserted = 0;
+    int interval = numSpots / 100;
+    
+    Map<String, ReporterProxy> preloaded = 
+      new HashMap<String, ReporterProxy>(numSpots);
+    
+    // Scan the file and create proxies for all reporters
+    File manifest = getManifestFile(dc, design);
+    String mfFilename = manifest.getName();
+    FlatFileParser ffp = getManifestParser(manifest);
+    Mapper reporterMapper = ffp.getMapper("\\IlmnID\\");
+    Mapper addressMapper = ffp.getMapper("\\AddressA_ID\\");
+    while (ffp.hasMoreData())
+    {
+      checkInterrupted();
+      currentSpot++;
+      FlatFileParser.Data dataLine = ffp.nextData();
+      String externalId = reporterMapper.getValue(dataLine);
+      Integer addressId = addressMapper.getInt(dataLine);
+      if (externalId == null || addressId == null) continue;
+      
+      ReporterProxy proxy = preloaded.get(externalId);
+      if (proxy == null)
+      {
+        proxy = new ReporterProxy();
+        preloaded.put(externalId, proxy);
+      }
+      proxy.addAddressId(addressId);
+
+      if (currentSpot % interval == 0 && progress != null)
+      {
+        progress.displayAbsolute(baseProgress + currentSpot, 
+          "Loading reporters from file '" + mfFilename + "': " + 
+          currentSpot + " of " + numSpots + " done.");
+      }
+    }
+    return preloaded;
+  }
+  
+  /**
+    Query the reporters table and fill in the Reporter.id value for
+    all reporters that have been loaded from the SNP manifest file.
+  */
+  private void mapReporterIds(DbControl dc, Map<String, ReporterProxy> preloaded,
+    SimpleAbsoluteProgressReporter progress)
+    throws SQLException
+  {
+    int numSpots = preloaded.size();
+    long baseProgress = progress == null ? 0 : progress.getCompleted();
     int currentSpot = 0;
     int interval = numSpots / 100;
-    float progressFactor = 100.0f / numSpots;
-    float[] snpdata = new float[3];
-    while (ffp.hasMoreData())
-    {         
+
+    // Fill the proxies with ID:s for the reporters
+    DynamicReporterQuery query = Reporter.getDynamicQuery();
+    query.select(Dynamic.selectReporter("id"));
+    query.select(Dynamic.selectReporter("externalId"));
+    DynamicResultIterator result = query.iterate(dc);
+    while (result.hasNext())
+    {
+      SqlResult reporter = result.next();
+      ReporterProxy proxy = preloaded.get(reporter.getString(2));
+      if (proxy != null)
+      {
+        checkInterrupted();
+        currentSpot++;
+        proxy.setTheId(reporter.getInt(1));
+        
+        if (currentSpot % interval == 0 && progress != null)
+        {
+          progress.displayAbsolute(baseProgress + currentSpot, 
+            "Loading reporters from database: " + 
+            currentSpot + " of " + numSpots + " done.");
+        }
+      }
+    }
+    result.close();
+    
+    if (progress != null) 
+    {
+      progress.displayAbsolute(baseProgress+numSpots, 
+        "Loading reporters from database: " + 
+        numSpots + " of " + numSpots + " done.");
+    }
+  }
+
+  /**
+    Insert position -> reporter mapping in the database. Data is taken
+    from the preloaded map. The returned set contains the addressId:s
+    of positions that has been skipped because no reporter was found.
+  */
+  private Set<Integer> insertPosData(DbControl dc, PositionBatcher posBatcher, 
+    Map<String, ReporterProxy> preloaded, SimpleAbsoluteProgressReporter progress)
+  {
+    // Insert pos -> reporter mapping into database
+    String missingReporter = getErrorOption(missingReporterErrorParameter.getName());
+    boolean skipLine = "skip".equals(missingReporter);
+    boolean useNull = "null".equals(missingReporter);
+    Set<Integer> skipped = new HashSet<Integer>();
+
+    int numSpots = preloaded.size();
+    long baseProgress = progress == null ? 0 : progress.getCompleted();
+    int currentSpot = 0;
+    int interval = numSpots / 100;
+    
+    for (Map.Entry<String, ReporterProxy> entry : preloaded.entrySet())
+    {
       checkInterrupted();
-      FlatFileParser.Data data = ffp.nextData();
-      snpdata[0] = gTypeMapper.getFloat(data);
-      snpdata[1] = bAlleleMapper.getFloat(data);
-      snpdata[2] = ratioMapper.getFloat(data);
-      int addressId = addressMapper.getInt(data);
-      spotBatcher.insert(baCol, addressId, snpdata);
+      currentSpot++;
+      ReporterProxy proxy = entry.getValue();
+      Integer addressId = proxy.getAddressId();
+      ReporterData reporter = proxy.getReporter();
+      
+      // Missing reporter?
+      if (reporter == null)
+      {
+        if (skipLine) 
+        {
+          skipped.add(addressId);
+          if (proxy.hasMultiple()) skipped.addAll(proxy.getMultipleAddressIds());
+          continue;
+        }
+        if (!useNull) throw new ItemNotFoundException("Reporter[id=" + entry.getKey() +"]");
+      }
+      posBatcher.insert(addressId, reporter);
+      if (proxy.hasMultiple())
+      {
+        for (int aId : proxy.getMultipleAddressIds())
+        {
+          posBatcher.insert(addressId, reporter);
+        }
+      }
       
       if (currentSpot % interval == 0 && progress != null)
       {
-        progress.display((int)(progressFactor * currentSpot), 
-          "Inserting spot data for raw bioassay '" + rba.getName() + "'. " + 
+        checkInterrupted();
+        progress.displayAbsolute(baseProgress + currentSpot, 
+          "Mapping reporters to positions: " + 
           currentSpot + " of " + numSpots + " done.");
-      }
-    }
-    spotBatcher.flush();
-
+      }     
+    }
+    return skipped;
   }