Static accessor methods for observation extensions, implementation for...

Static accessor methods for observation extensions, implementation for patient, visit. i2b2VisitStore incomplete

HIStream-i2b2/src/main/java/de/sekmi/histream/i2b2/PostgresPatientStore.java

@@ -584,22 +584,20 @@ public class PostgresPatientStore extends PostgresExtension<I2b2Patient> impleme
 		insertIde.setString(6, source.getSourceId());
 		insertIde.executeUpdate();
 	}
-	@Override
-	public I2b2Patient createInstance(Observation fact) {
-
-		
-		I2b2Patient pat = getCached(fact.getPatientId());
+	
+	private I2b2Patient getOrCreateInstance(String patientId, ExternalSourceType source){
+		I2b2Patient pat = getCached(patientId);
 		if( pat == null ){
 			// string id not known to cache
 			// create new patient
 			maxPatientNum ++;
 			int num = maxPatientNum;
 			pat = new I2b2Patient(num);
-			pat.setId(fact.getPatientId());
+			pat.setId(patientId);
 			
 			
 			// don't use source metadata, since we only know the patient id
-			pat.setSourceId(fact.getSourceId());
+			pat.setSourceId(source.getSourceId());
 			//pat.setSourceTimestamp(fact.getSourceTimestamp());
 
 			// put in cache and insert into storage
@@ -609,7 +607,7 @@ public class PostgresPatientStore extends PostgresExtension<I2b2Patient> impleme
 				insertPatient(pat);
 				
 				// insert ide into patient_mapping
-				insertIde(num, pat.getId(), "A", fact);
+				insertIde(num, pat.getId(), "A", source);
 			} catch (SQLException e) {
 				insertionException(pat, e);
 			} 
@@ -620,6 +618,11 @@ public class PostgresPatientStore extends PostgresExtension<I2b2Patient> impleme
 		}
 		return pat;
 	}
+	
+	@Override
+	public I2b2Patient createInstance(Observation fact) {
+		return getOrCreateInstance(fact.getPatientId(), fact);		
+	}
 
 	@Override
 	public Class<?>[] getInstanceTypes() {
@@ -627,12 +630,15 @@ public class PostgresPatientStore extends PostgresExtension<I2b2Patient> impleme
 	}
 
 	@Override
-	public I2b2Patient createInstance() throws UnsupportedOperationException {
-		throw new UnsupportedOperationException();
+	public I2b2Patient createInstance(Object... args) throws IllegalArgumentException {
+		if( args.length != 2 || !(args[0] instanceof String) || !(args[1] instanceof ExternalSourceType) ){
+			throw new IllegalArgumentException("Expected arguments: String patientId, ExternalSourceType source");
+		}
+		return getOrCreateInstance((String)args[0], (ExternalSourceType)args[1]);
 	}
 
 	@Override
-	public Patient retrieve(String id) {
+	public I2b2Patient retrieve(String id) {
 		return idCache.get(id);
 	}
 

HIStream-i2b2/src/main/java/de/sekmi/histream/i2b2/PostgresVisitStore.java

@@ -613,8 +613,9 @@ public class PostgresVisitStore extends PostgresExtension<I2b2Visit>{
 	}
 
 	@Override
-	public I2b2Visit createInstance() throws UnsupportedOperationException {
-		throw new UnsupportedOperationException();
+	public I2b2Visit createInstance(Object... args) throws UnsupportedOperationException {
+		// TODO: implement
+		throw new UnsupportedOperationException("TODO: implement");
 	}
 
 	@Override
@@ -633,4 +634,5 @@ public class PostgresVisitStore extends PostgresExtension<I2b2Visit>{
 		if( count != 0 )log.info("Updated "+count+" visits in database");
 	}
 
+
 }

histream-core/src/main/java/de/sekmi/histream/Extension.java

@@ -48,8 +48,9 @@ public interface Extension<T>{
 	 * calls to this method will result in an {@link UnsupportedOperationException}.
 	 * @return new instance 
 	 * @throws UnsupportedOperationException if instance creation without {@link Observation} is not possible.
+	 * @throws IllegalArgumentException if given arguments are unsuitable to instantiate/identify this type.
 	 */
-	T createInstance() throws UnsupportedOperationException;
+	T createInstance(Object... args) throws UnsupportedOperationException, IllegalArgumentException;
 	
 	/**
 	 * Get class of the instance type. Should be a basic interface like Patient, Visit, Location, Concept, etc.

histream-core/src/main/java/de/sekmi/histream/ExtensionAccessor.java

@@ -40,6 +40,22 @@ public interface ExtensionAccessor<T> {
 	 */
 	T access(Observation observation);
 	
+	/**
+	 * Sets the extension type instance for the given observation.
+	 * @param observation observation to set the instance for
+	 * @param ext extension instance
+	 */
 	void set(Observation observation, T ext);
 	// TODO: if necessary, create method isAvailable which does not create the instance automatically
+	
+	/**
+	 * Get an extension type instance which is independent of a single observation. 
+	 * The instance is identified by the provided arguments.
+	 * 
+	 * @param args arguments to identify the static instance
+	 * @return extension instance
+	 * @throws UnsupportedOperationException if this extension does not support static instances
+	 * @throws IllegalArgumentException if the arguments are inappropriate to identify instances of this type.
+	 */
+	T accessStatic(Object... args) throws UnsupportedOperationException, IllegalArgumentException;
 }

histream-core/src/main/java/de/sekmi/histream/ObservationFactory.java

@@ -26,6 +26,8 @@ package de.sekmi.histream;
  * Single instance which generates all observations. 
  * Manages extensions which enhance/annotate observations.
  * 
+ * TODO replace interface with implementation - one implementation is sufficient.
+ * 
  * @author marap1
  *
  */
@@ -60,4 +62,6 @@ public interface ObservationFactory {
 	 * @return new observation
 	 */
 	Observation createObservation(String patientId, String conceptId, DateTimeAccuracy startTime);
+	
+	
 }

histream-core/src/main/java/de/sekmi/histream/impl/ExternalSourceImpl.java

+package de.sekmi.histream.impl;
+
+import java.time.Instant;
+
+import de.sekmi.histream.ext.ExternalSourceType;
+
+public class ExternalSourceImpl implements ExternalSourceType {
+	
+	protected Instant sourceTimestamp;
+	protected String sourceId;
+
+
+	@Override
+	public Instant getSourceTimestamp() {
+		return sourceTimestamp;
+	}
+
+	@Override
+	public void setSourceTimestamp(Instant instant) {
+		this.sourceTimestamp = instant;
+	}
+
+	@Override
+	public String getSourceId() {
+		return sourceId;
+	}
+
+	@Override
+	public void setSourceId(String sourceSystemId) {
+		this.sourceId = sourceSystemId;
+	}
+}

histream-core/src/main/java/de/sekmi/histream/impl/ObservationFactoryImpl.java

@@ -175,6 +175,11 @@ public class ObservationFactoryImpl implements ObservationFactory{
 			public void set(Observation observation, T ext) {
 				setExtension((ObservationImpl)observation, entry, ext);
 			}
+
+			@Override
+			public T accessStatic(Object... args) throws UnsupportedOperationException, IllegalArgumentException {
+				return null;
+			}
 		};
 	}
 

histream-core/src/main/java/de/sekmi/histream/impl/ObservationImpl.java

@@ -21,7 +21,6 @@ package de.sekmi.histream.impl;
  */
 
 
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -51,7 +50,7 @@ import de.sekmi.histream.Value;
 @XmlAccessorType(XmlAccessType.NONE)
 @XmlType(propOrder={"abstractValue","modifierList"})
 @XmlSeeAlso({StringValue.class,NumericValue.class})
-public class ObservationImpl implements Observation{
+public class ObservationImpl extends ExternalSourceImpl implements Observation{
 	public static final String XML_NAMESPACE="http://sekmi.de/histream/ns/eav-data";
 	@XmlTransient
 	protected ObservationFactoryImpl factory;
@@ -80,9 +79,6 @@ public class ObservationImpl implements Observation{
 	@XmlAttribute(name="end")
 	protected DateTimeAccuracy endTime;
 
-	protected Instant sourceTimestamp;
-	protected String sourceId;
-
 	/**
 	 * Modifiers
 	 */
@@ -185,6 +181,10 @@ public class ObservationImpl implements Observation{
 	@Override
 	public ObservationFactory getFactory() {return factory;}
 
+	public void setFactory(ObservationFactoryImpl factory){
+		this.factory = factory;
+	}
+	
 	@Override
 	public <T> T getExtension(Class<T> extensionType) {
 		// delegate to factory
@@ -197,25 +197,6 @@ public class ObservationImpl implements Observation{
 		factory.setExtension(this, extensionType, extension);
 	}
 
-	@Override
-	public Instant getSourceTimestamp() {
-		return sourceTimestamp;
-	}
-
-	@Override
-	public void setSourceTimestamp(Instant instant) {
-		this.sourceTimestamp = instant;
-	}
-
-	@Override
-	public String getSourceId() {
-		return sourceId;
-	}
-
-	@Override
-	public void setSourceId(String sourceSystemId) {
-		this.sourceId = sourceSystemId;
-	}
 
 	@Override
 	public Modifier addModifier(String modifierId, Value value) {

histream-core/src/main/java/de/sekmi/histream/impl/SimplePatientExtension.java

@@ -32,7 +32,10 @@ public class SimplePatientExtension implements Extension<PatientImpl>{
 	public Class<?>[] getInstanceTypes() {return TYPES;}
 
 	@Override
-	public PatientImpl createInstance() {return new PatientImpl();}
+	public PatientImpl createInstance(Object... args) {
+		// TODO use string patient id as identification
+		return new PatientImpl();
+	}
 
 	@Override
 	public PatientImpl createInstance(Observation observation) {

histream-core/src/main/java/de/sekmi/histream/impl/SimpleVisitExtension.java

@@ -32,7 +32,8 @@ public class SimpleVisitExtension implements Extension<VisitImpl>{
 	public Class<?>[] getInstanceTypes() {return TYPES;}
 
 	@Override
-	public VisitImpl createInstance() {
+	public VisitImpl createInstance(Object... args) {
+		// TODO: use patient id and visit id to identify visit
 		return new VisitImpl();
 	}
 

histream-core/src/main/java/de/sekmi/histream/impl/XPathEvaluator.java

@@ -10,8 +10,6 @@ import javax.xml.bind.JAXBContext;
 import javax.xml.bind.JAXBException;
 import javax.xml.bind.Marshaller;
 import javax.xml.namespace.NamespaceContext;
-import javax.xml.parsers.DocumentBuilderFactory;
-import javax.xml.parsers.ParserConfigurationException;
 import javax.xml.transform.dom.DOMResult;
 import javax.xml.xpath.XPath;
 import javax.xml.xpath.XPathConstants;
@@ -20,8 +18,6 @@ import javax.xml.xpath.XPathConstants;
 import javax.xml.xpath.XPathExpressionException;
 import javax.xml.xpath.XPathFactory;
 
-import org.w3c.dom.DOMConfiguration;
-import org.w3c.dom.Document;
 import org.w3c.dom.Node;
 
 import de.sekmi.histream.Observation;

histream-core/src/main/java/de/sekmi/histream/io/JAXBObservationSupplier.java

@@ -3,6 +3,8 @@ package de.sekmi.histream.io;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
+import java.time.Instant;
+import java.util.Calendar;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -15,9 +17,14 @@ import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
 
 import de.sekmi.histream.DateTimeAccuracy;
+import de.sekmi.histream.ExtensionAccessor;
 import de.sekmi.histream.Observation;
 import de.sekmi.histream.ObservationFactory;
 import de.sekmi.histream.ObservationSupplier;
+import de.sekmi.histream.ext.Patient;
+import de.sekmi.histream.ext.Visit;
+import de.sekmi.histream.impl.ExternalSourceImpl;
+import de.sekmi.histream.impl.ObservationFactoryImpl;
 import de.sekmi.histream.impl.ObservationImpl;
 
 public class JAXBObservationSupplier implements ObservationSupplier {
@@ -29,6 +36,12 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 	
 
 	private ObservationFactory factory;
+	private ExtensionAccessor<Patient> patientAccessor;
+	private ExtensionAccessor<Visit> visitAccessor;
+	private Patient currentPatient;
+	private Visit currentVisit;
+	private ExternalSourceImpl source;
+	
 	private Unmarshaller unmarshaller;
 	private XMLStreamReader reader;
 	private Map<String,String> meta;
@@ -57,6 +70,10 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 		// TODO: set schema
 		//unmarshaller.setSchema(schema);
 		this.reader = reader;
+		this.source = new ExternalSourceImpl();
+		this.visitAccessor = factory.getExtensionAccessor(Visit.class);
+		this.patientAccessor = factory.getExtensionAccessor(Patient.class);
+		
 		readToRoot();
 		readMeta();
 		readPatient();
@@ -74,6 +91,10 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 		reader.nextTag();
 	}
 	
+	private Instant parseSourceTimestamp(String timestamp){
+		Calendar cal = javax.xml.bind.DatatypeConverter.parseDateTime(timestamp);
+		return cal.toInstant();
+	}
 	private void readMeta()throws XMLStreamException{
 		if( !reader.isStartElement() || !reader.getLocalName().equals("meta") )return;
 		
@@ -88,9 +109,15 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 			reader.nextTag();
 		}
 		if( reader.getLocalName().equals("source") ){
-			meta.put("source.timestamp", reader.getAttributeValue(null, "timestamp"));
-			meta.put("source.id", reader.getAttributeValue(null, "id"));
-
+			String sourceTimestamp = reader.getAttributeValue(null, "timestamp");
+			meta.put("source.timestamp", sourceTimestamp);
+			String sourceId = reader.getAttributeValue(null, "id");
+			meta.put("source.id", sourceId);
+			// set source
+			source.setSourceId(sourceId);
+			
+			source.setSourceTimestamp(parseSourceTimestamp(sourceTimestamp));
+			
 			reader.nextTag();
 			// should be end element
 			reader.nextTag();
@@ -121,6 +148,9 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 			patientData.put(reader.getLocalName(), reader.getElementText());
 			reader.nextTag();
 		}
+		// register with extension
+		currentPatient = patientAccessor.accessStatic(patientId, source);
+		// TODO set patient data
 	}
 	/**
 	 * Reads encounter element with content. 
@@ -149,6 +179,8 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 		}
 		// TODO assert at <facts>
 		reader.nextTag();
+		
+		// TODO register visit with extension, set visit data
 	}
 	protected Observation readObservation()throws XMLStreamException{
 		if( reader.isWhiteSpace() ){
@@ -210,7 +242,12 @@ public class JAXBObservationSupplier implements ObservationSupplier {
 		}
 		// TODO set etc. from visit
 		
-		// TODO set ObservationFactory, initialize extensions
+		// set ObservationFactory, initialize extensions
+		fact.setFactory((ObservationFactoryImpl)factory);
+		patientAccessor.set(fact, currentPatient);
+		// TODO set current visit
+		// TODO add tests to verify patient/visit extensions for parsed facts
+		
 		return fact;
 	}
 

histream-core/src/test/java/de/sekmi/histream/io/FileObservationProviderTest.java

@@ -44,6 +44,8 @@ import de.sekmi.histream.Observation;
 import de.sekmi.histream.ObservationFactory;
 import de.sekmi.histream.ObservationSupplier;
 import de.sekmi.histream.Value;
+import de.sekmi.histream.ext.Patient;
+import de.sekmi.histream.ext.Visit;
 import de.sekmi.histream.Modifier;
 import de.sekmi.histream.impl.ObservationFactoryImpl;
 import de.sekmi.histream.impl.SimplePatientExtension;
@@ -65,6 +67,7 @@ public class FileObservationProviderTest {
 		factory.registerExtension(new SimplePatientExtension());
 		factory.registerExtension(new SimpleVisitExtension());
 		//factory.registerExtension(new ConceptExtension());
+		
 	}	
 	
 	@Before
@@ -74,6 +77,13 @@ public class FileObservationProviderTest {
 				Assert.assertEquals("T:date:secs", o.getConceptId());
 				Assert.assertEquals(ChronoUnit.SECONDS, o.getStartTime().getAccuracy());
 				Assert.assertEquals(3, o.getStartTime().getLong(ChronoField.SECOND_OF_MINUTE));
+				Patient p = o.getExtension(Patient.class);
+				Assert.assertNotNull("Patient extension required", p);
+				Assert.assertEquals("XX12345", p.getId());
+				// TODO: test more patient information
+				Visit v = o.getExtension(Visit.class);
+				Assert.assertNotNull("Visit extension required", v);
+				// TODO test visit information
 			},
 			(Observation o) ->  {
 				Assert.assertEquals("T:date:mins", o.getConceptId());

histream-import/src/main/java/de/sekmi/histream/etl/FactGroupingQueue.java

@@ -74,11 +74,13 @@ public class FactGroupingQueue{
 		tableIndex = 0;
 		
 		currentPatient = patientTable.get();
+		// TODO sync patient with extension factory
 		addFactsToWorkQueue(currentPatient);
 		
 		// for every patient, facts without visitId (=null) are parsed first
 		currentVisitId = null;		
 		nextVisit = visitTable.get();
+		// TODO sync visit with extension factory
 	}
 	
 	private void addFactsToWorkQueue(FactRow r){
@@ -122,8 +124,9 @@ public class FactGroupingQueue{
 			// no more fitting facts in current prefetched rows
 			// try to get next visit for current patient
 			if( nextVisit != null && nextVisit.getPatientId().equals(currentPatient.getPatientId()) ){
-				// next visit also belongs to currrent patient, continue
+				// next visit also belongs to current patient, continue
 				currentVisitId = nextVisit.getVisitId();
+				// TODO: sync visit with extension factory
 				addFactsToWorkQueue(nextVisit);
 				nextVisit = visitTable.get();
 				tableIndex = 0;
@@ -138,6 +141,7 @@ public class FactGroupingQueue{
 					// we are done
 					break;
 				}
+				// TODO: sync patient with extension factory
 				addFactsToWorkQueue(currentPatient);
 				tableIndex = 0;
 				// goto top

histream-import/src/test/java/de/sekmi/histream/etl/TestETLSupplier.java

@@ -39,6 +39,7 @@ public class TestETLSupplier {
 			debug_str.append(fact.getValue());
 			
 			System.out.println(debug_str.toString());
+			// TODO test patient extension, visit extension
 		}
 		s.close();
 	}