Merge pull request #1396 from Yaqiang/develop

Enhance Bufr data support including multi-category messages

Author: lesserwhirls

bufr/src/main/java/ucar/nc2/iosp/bufr/BufrConfig.java

@@ -45,6 +45,7 @@ static BufrConfig openFromMessage(RandomAccessFile raf, Message m, Element iospP
   }
 
   private String filename;
+  private Message message;
   private StandardFields.StandardFieldsFromMessage standardFields;
   private FieldConverter rootConverter;
   private int messHash;
@@ -86,6 +87,7 @@ private BufrConfig(RandomAccessFile raf) {
 
   private BufrConfig(RandomAccessFile raf, Message m) throws IOException {
     this.filename = raf.getLocation();
+    this.message = m;
     this.messHash = m.hashCode();
     this.rootConverter = new FieldConverter(m.ids.getCenterId(), m.getRootDataDescriptor());
     standardFields = StandardFields.extract(m);
@@ -95,6 +97,10 @@ public String getFilename() {
     return filename;
   }
 
+  public Message getMessage() {
+    return this.message;
+  }
+
   public FieldConverter getRootConverter() {
     return rootConverter;
   }

bufr/src/main/java/ucar/nc2/iosp/bufr/BufrIosp2.java

@@ -5,10 +5,8 @@
 package ucar.nc2.iosp.bufr;
 
 import java.io.IOException;
-import java.util.Formatter;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
+import java.util.*;
+
 import org.jdom2.Element;
 import ucar.ma2.Array;
 import ucar.ma2.ArraySequence;
@@ -46,9 +44,11 @@ public static void setDebugFlags(ucar.nc2.util.DebugFlags debugFlag) {
     debugIter = debugFlag.isSet("Bufr/iter");
   }
 
-  private Structure obsStructure;
-  private Message protoMessage; // prototypical message: all messages in the file must be the same.
+  // private Structure obsStructure;
+  // private Message protoMessage; // prototypical message: all messages in the file must be the same.
   private MessageScanner scanner;
+  private List<Message> protoMessages; // prototypical messages: the messages with different category.
+  private List<RootVariable> rootVariables;
   private HashSet<Integer> messHash;
   private boolean isSingle;
   private BufrConfig config;
@@ -69,25 +69,57 @@ public void build(RandomAccessFile raf, Group.Builder rootGroup, CancelTask canc
     super.open(raf, rootGroup.getNcfile(), cancelTask);
 
     scanner = new MessageScanner(raf);
-    protoMessage = scanner.getFirstDataMessage();
+    Message protoMessage = scanner.getFirstDataMessage();
     if (protoMessage == null)
       throw new IOException("No data messages in the file= " + raf.getLocation());
     if (!protoMessage.isTablesComplete())
       throw new IllegalStateException("BUFR file has incomplete tables");
 
+    // get all prototype messages - contains different message category in a Bufr data file
+    protoMessages = new ArrayList<>();
+    protoMessages.add(protoMessage);
+    int category = protoMessage.ids.getCategory();
+    while (scanner.hasNext()) {
+      Message message = scanner.next();
+      if (message.ids.getCategory() != category) {
+        protoMessages.add(message);
+        category = message.ids.getCategory();
+      }
+    }
+
     // just get the fields
     config = BufrConfig.openFromMessage(raf, protoMessage, iospParam);
 
     // this fills the netcdf object
-    new BufrIospBuilder(protoMessage, config, rootGroup, raf.getLocation());
+    if (this.protoMessages.size() == 1) {
+      new BufrIospBuilder(protoMessage, config, rootGroup, raf.getLocation());
+    } else {
+      List<BufrConfig> configs = new ArrayList<>();
+      for (Message message : protoMessages) {
+        configs.add(BufrConfig.openFromMessage(raf, message, iospParam));
+      }
+      new BufrIospBuilder(protoMessage, configs, rootGroup, raf.getLocation());
+    }
     isSingle = false;
   }
 
   @Override
   public void buildFinish(NetcdfFile ncfile) {
-    obsStructure = (Structure) ncfile.findVariable(obsRecordName);
-    // The proto DataDescriptor must have a link to the Sequence object to read nested Sequences.
-    connectSequences(obsStructure.getVariables(), protoMessage.getRootDataDescriptor().getSubKeys());
+    // support multiple root variables in one Bufr data file
+    this.rootVariables = new ArrayList<>();
+    if (this.protoMessages.size() == 1) {
+      Structure obsStructure = (Structure) ncfile.findVariable(obsRecordName);
+      // The proto DataDescriptor must have a link to the Sequence object to read nested Sequences.
+      connectSequences(obsStructure.getVariables(), protoMessages.get(0).getRootDataDescriptor().getSubKeys());
+      this.rootVariables.add(new RootVariable(protoMessages.get(0), obsStructure));
+    } else {
+      for (int i = 0; i < this.protoMessages.size(); i++) {
+        Structure variable = (Structure) ncfile.getVariables().get(i);
+        Message message = protoMessages.get(i);
+        connectSequences(variable.getVariables(), message.getRootDataDescriptor().getSubKeys());
+        this.rootVariables.add(new RootVariable(message, variable));
+      }
+    }
   }
 
   private void connectSequences(List<Variable> variables, List<DataDescriptor> dataDescriptors) {
@@ -116,27 +148,48 @@ public void open(RandomAccessFile raf, NetcdfFile ncfile, CancelTask cancelTask)
     super.open(raf, ncfile, cancelTask);
 
     scanner = new MessageScanner(raf);
-    protoMessage = scanner.getFirstDataMessage();
+    Message protoMessage = scanner.getFirstDataMessage();
     if (protoMessage == null)
       throw new IOException("No data messages in the file= " + ncfile.getLocation());
     if (!protoMessage.isTablesComplete())
       throw new IllegalStateException("BUFR file has incomplete tables");
 
+    // get all prototype messages - contains different message category in a Bufr data file
+    protoMessages = new ArrayList<>();
+    protoMessages.add(protoMessage);
+    int category = protoMessage.ids.getCategory();
+    while (scanner.hasNext()) {
+      Message message = scanner.next();
+      if (message.ids.getCategory() != category) {
+        protoMessages.add(message);
+        category = message.ids.getCategory();
+      }
+    }
+
     // just get the fields
     config = BufrConfig.openFromMessage(raf, protoMessage, iospParam);
 
     // this fills the netcdf object
-    Construct2 construct = new Construct2(protoMessage, config, ncfile);
-    obsStructure = construct.getObsStructure();
+    if (this.protoMessages.size() == 1) {
+      Construct2 construct = new Construct2(protoMessage, config, ncfile);
+    } else {
+      List<BufrConfig> configs = new ArrayList<>();
+      for (Message message : protoMessages) {
+        configs.add(BufrConfig.openFromMessage(raf, message, iospParam));
+      }
+      Construct2 construct = new Construct2(protoMessage, configs, ncfile);
+    }
+
     ncfile.finish();
+    buildFinish(ncfile);
     isSingle = false;
   }
 
   // for BufrMessageViewer
   public void open(RandomAccessFile raf, NetcdfFile ncfile, Message single) throws IOException {
     this.raf = raf;
 
-    protoMessage = single;
+    Message protoMessage = single;
     protoMessage.getRootDataDescriptor(); // construct the data descriptors, check for complete tables
     if (!protoMessage.isTablesComplete())
       throw new IllegalStateException("BUFR file has incomplete tables");
@@ -145,7 +198,7 @@ public void open(RandomAccessFile raf, NetcdfFile ncfile, Message single) throws
 
     // this fills the netcdf object
     Construct2 construct = new Construct2(protoMessage, config, ncfile);
-    obsStructure = construct.getObsStructure();
+    Structure obsStructure = construct.getObsStructure();
     isSingle = true;
 
     ncfile.finish();
@@ -175,28 +228,67 @@ public Element getElem() {
 
   @Override
   public Array readData(Variable v2, Section section) {
-    findRootSequence();
-    return new ArraySequence(obsStructure.makeStructureMembers(), new SeqIter(), nelems);
+    RootVariable rootVariable = findRootSequence(v2);
+    Structure obsStructure = rootVariable.getVariable();
+    return new ArraySequence(obsStructure.makeStructureMembers(), new SeqIter(rootVariable), nelems);
   }
 
   @Override
   public StructureDataIterator getStructureIterator(Structure s, int bufferSize) {
-    findRootSequence();
-    return isSingle ? new SeqIterSingle() : new SeqIter();
+    RootVariable rootVariable = findRootSequence(s);
+    return isSingle ? new SeqIterSingle(rootVariable) : new SeqIter(rootVariable);
+  }
+
+  private Structure findRootSequence() {
+    return (Structure) this.ncfile.findVariable(BufrIosp2.obsRecordName);
   }
 
-  private void findRootSequence() {
-    this.obsStructure = (Structure) this.ncfile.findVariable(BufrIosp2.obsRecordName);
+  // find root sequence from root variable list
+  private RootVariable findRootSequence(Variable var) {
+    for (RootVariable rootVariable : this.rootVariables) {
+      if (rootVariable.getVariable().getShortName().equals(var.getShortName())) {
+        return rootVariable;
+      }
+    }
+    return null;
+  }
+
+  // root variable contains prototype message and corresponding variable
+  private class RootVariable {
+    private Message protoMessage;
+    private Structure variable;
+
+    public RootVariable(Message message, Structure variable) {
+      this.protoMessage = message;
+      this.variable = variable;
+    }
+
+    public Message getProtoMessage() {
+      return this.protoMessage;
+    }
+
+    public Structure getVariable() {
+      return this.variable;
+    }
   }
 
   private class SeqIter implements StructureDataIterator {
     StructureDataIterator currIter;
     int recnum;
+    // add its own prototype message and observation structure
+    Message protoMessage;
+    Structure obsStructure;
 
-    SeqIter() {
+    SeqIter(Message message, Structure structure) {
+      this.protoMessage = message;
+      this.obsStructure = structure;
       reset();
     }
 
+    SeqIter(RootVariable rootVariable) {
+      this(rootVariable.protoMessage, rootVariable.variable);
+    }
+
     @Override
     public StructureDataIterator reset() {
       recnum = 0;
@@ -286,11 +378,20 @@ public void close() {
   private class SeqIterSingle implements StructureDataIterator {
     StructureDataIterator currIter;
     int recnum;
+    // add its own prototype message and observation structure
+    Message protoMessage;
+    Structure obsStructure;
 
-    SeqIterSingle() {
+    SeqIterSingle(Message message, Structure structure) {
+      protoMessage = message;
+      obsStructure = structure;
       reset();
     }
 
+    SeqIterSingle(RootVariable rootVariable) {
+      this(rootVariable.protoMessage, rootVariable.variable);
+    }
+
     @Override
     public StructureDataIterator reset() {
       recnum = 0;
@@ -350,7 +451,7 @@ public void close() {
   public String getDetailInfo() {
     Formatter ff = new Formatter();
     ff.format("%s", super.getDetailInfo());
-    protoMessage.dump(ff);
+    protoMessages.get(0).dump(ff);
     ff.format("%n");
     config.show(ff);
     return ff.toString();

bufr/src/main/java/ucar/nc2/iosp/bufr/BufrIospBuilder.java

@@ -30,7 +30,7 @@ class BufrIospBuilder {
   private static final boolean warnUnits = false;
 
   private final Group.Builder rootGroup;
-  private final Sequence.Builder recordStructure;
+  private Sequence.Builder recordStructure;
   private final Formatter coordinates = new Formatter();
 
   private int tempNo = 1; // fishy
@@ -75,6 +75,42 @@ class BufrIospBuilder {
     }
   }
 
+  BufrIospBuilder(Message proto, List<BufrConfig> bufrConfigs, Group.Builder root, String location) {
+    this.rootGroup = root;
+
+    // global Attributes
+    AttributeContainerMutable atts = root.getAttributeContainer();
+    atts.addAttribute(CDM.HISTORY, "Read using CDM BufrIosp2");
+    atts.addAttribute("location", location);
+
+    atts.addAttribute("BUFR:categoryName", proto.getLookup().getCategoryName());
+    atts.addAttribute("BUFR:subCategoryName", proto.getLookup().getSubCategoryName());
+    atts.addAttribute("BUFR:centerName", proto.getLookup().getCenterName());
+    atts.addAttribute(BufrIosp2.centerId, proto.ids.getCenterId());
+    atts.addAttribute("BUFR:subCenter", proto.ids.getSubCenterId());
+    atts.addAttribute("BUFR:table", proto.ids.getMasterTableId());
+    atts.addAttribute("BUFR:tableVersion", proto.ids.getMasterTableVersion());
+    atts.addAttribute("BUFR:localTableVersion", proto.ids.getLocalTableVersion());
+    atts.addAttribute("Conventions", "BUFR/CDM");
+    atts.addAttribute("BUFR:edition", proto.is.getBufrEdition());
+
+    String header = proto.getHeader();
+    if (header != null && !header.isEmpty()) {
+      atts.addAttribute("WMO Header", header);
+    }
+
+    for (BufrConfig bufrConfig : bufrConfigs) {
+      String varName = proto.getLookup().getCategoryName(bufrConfig.getMessage().ids.getCategory());
+      Sequence.Builder rs = Sequence.builder().setName(varName);
+      this.rootGroup.addVariable(rs);
+      makeObsRecord(bufrConfig, rs);
+      String coordS = coordinates.toString();
+      if (!coordS.isEmpty()) {
+        rs.addAttribute(new Attribute("coordinates", coordS));
+      }
+    }
+  }
+
   Sequence.Builder getObsStructure() {
     return recordStructure;
   }
@@ -117,6 +153,44 @@ private void makeObsRecord(BufrConfig bufrConfig) {
     }
   }
 
+  private void makeObsRecord(BufrConfig bufrConfig, Sequence.Builder rs) {
+    BufrConfig.FieldConverter root = bufrConfig.getRootConverter();
+    for (BufrConfig.FieldConverter fld : root.flds) {
+      DataDescriptor dkey = fld.dds;
+      if (!dkey.isOkForVariable()) {
+        continue;
+      }
+
+      if (dkey.replication == 0) {
+        addSequence(rootGroup, rs, fld);
+
+      } else if (dkey.replication > 1) {
+
+        List<BufrConfig.FieldConverter> subFlds = fld.flds;
+        List<DataDescriptor> subKeys = dkey.subKeys;
+        if (subKeys.size() == 1) { // only one member
+          DataDescriptor subDds = dkey.subKeys.get(0);
+          BufrConfig.FieldConverter subFld = subFlds.get(0);
+          if (subDds.dpi != null) {
+            addDpiStructure(rs, fld, subFld);
+
+          } else if (subDds.replication == 1) { // one member not a replication
+            Variable.Builder v = addVariable(rootGroup, rs, subFld, dkey.replication);
+            v.setSPobject(fld); // set the replicating field as SPI object
+
+          } else { // one member is a replication (two replications in a row)
+            addStructure(rootGroup, rs, fld, dkey.replication);
+          }
+        } else if (subKeys.size() > 1) {
+          addStructure(rootGroup, rs, fld, dkey.replication);
+        }
+
+      } else { // replication == 1
+        addVariable(rootGroup, rs, fld, dkey.replication);
+      }
+    }
+  }
+
   private void addStructure(Group.Builder group, Structure.Builder parent, BufrConfig.FieldConverter fld, int count) {
     DataDescriptor dkey = fld.dds;
     String uname = findUniqueName(parent, fld.getName(), "struct");

bufr/src/main/java/ucar/nc2/iosp/bufr/BufrTableLookup.java

@@ -127,11 +127,14 @@ public String getSubCategoryName() { // throws IOException {
     return subcatName;
   }
 
-
   public String getCategoryName() {
     return TableA.getDataCategory(getCategory());
   }
 
+  public String getCategoryName(int cat) {
+    return TableA.getDataCategoryName(cat);
+  }
+
   public String getCategoryNo() {
     String result = getCategory() + "." + getSubCategory();
     if (getLocalSubCategory() >= 0)