ITQC/gls_docs
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ATEST-442 -> Extrema

Browse Source
This commit is contained in:
Fabian Märki
2016-06-10 17:43:34 +02:00
parent 6638043102
commit fb327f7457
7 changed files with 457 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java
+2
View File
@@ -32,6 +32,7 @@ import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.operation.Response;
import ch.psi.daq.domain.query.operation.aggregation.extrema.AbstractExtremaMeta;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.analyzer.QueryAnalyzerImpl;
import ch.psi.daq.query.config.QueryConfig;
@@ -86,6 +87,7 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
// won't. This way, the user can specify which columns are to be received.
objectMapper.addMixIn(DataEvent.class, PropertyFilterMixin.class);
objectMapper.addMixIn(Statistics.class, PropertyFilterMixin.class);
objectMapper.addMixIn(AbstractExtremaMeta.class, PropertyFilterMixin.class);
objectMapper.addMixIn(EnumMap.class, PropertyFilterMixin.class);
objectMapper.addMixIn(Response.class, PolymorphicResponseMixIn.class);
src/main/java/ch/psi/daq/queryrest/response/csv/CSVResponseStreamWriter.java
+44 -6
View File
@@ -35,6 +35,7 @@ import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.model.Query;
@@ -47,11 +48,12 @@ import ch.psi.daq.queryrest.response.ResponseStreamWriter;
*/
public class CSVResponseStreamWriter implements ResponseStreamWriter {
private static final Logger LOGGER = LoggerFactory.getLogger(CSVResponseStreamWriter.class);
public static final char DELIMITER_CVS = ';';
public static final String DELIMITER_ARRAY = ",";
public static final char DELIMITER_CHANNELNAME_FIELDNAME = '.';
public static final String EMPTY_VALUE = "";
public static final String FIELDNAME_EXTREMA = "extrema";
private static final Function<Pair<ChannelName, DataEvent>, ChannelName> KEY_PROVIDER = (pair) -> pair.getKey();
// try to match sync data (bsread) with non sync data (epics) based on the time usin 10 millis
// buckets.
@@ -62,7 +64,8 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
@Override
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results, final OutputStream out) throws Exception {
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
final OutputStream out) throws Exception {
AtomicReference<Exception> exception = new AtomicReference<>();
final Map<ChannelName, Stream<Pair<ChannelName, DataEvent>>> streams = new LinkedHashMap<>(results.size());
@@ -151,14 +154,20 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
private void setupChannelColumns(DAQQueryElement daqQuery, BackendQuery backendQuery, ChannelName channelName,
Collection<String> header, Collection<Pair<ChannelName, Function<DataEvent, String>>> accessors) {
Set<QueryField> queryFields = daqQuery.getFields();
List<Aggregation> aggregations = daqQuery.getAggregation() != null ? daqQuery.getAggregation().getAggregations() : null;
List<Aggregation> aggregations =
daqQuery.getAggregation() != null ? daqQuery.getAggregation().getAggregations() : null;
List<Extrema> extrema = daqQuery.getAggregation() != null ? daqQuery.getAggregation().getExtrema() : null;
QueryAnalyzer queryAnalyzer = queryAnalizerFactory.apply(backendQuery);
for (QueryField field : queryFields) {
if (!(QueryField.value.equals(field) && queryAnalyzer.isAggregationEnabled())) {
header.add(channelName.getName() + DELIMITER_CHANNELNAME_FIELDNAME + field.name());
StringBuilder buf = new StringBuilder(3)
.append(channelName.getName())
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(field.name());
header.add(buf.toString());
accessors.add(Pair.of(channelName, new QueryFieldStringifyer(field.getAccessor(), EMPTY_VALUE,
DELIMITER_ARRAY)));
}
@@ -166,10 +175,39 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
if (aggregations != null && queryAnalyzer.isAggregationEnabled()) {
for (Aggregation aggregation : aggregations) {
header.add(channelName.getName() + DELIMITER_CHANNELNAME_FIELDNAME + QueryField.value.name()
+ DELIMITER_CHANNELNAME_FIELDNAME + aggregation.name());
StringBuilder buf = new StringBuilder(5)
.append(channelName.getName())
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(QueryField.value.name())
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(aggregation.name());
header.add(buf.toString());
accessors.add(Pair.of(channelName, new AggregationStringifyer(aggregation.getAccessor(), EMPTY_VALUE)));
}
}
if (extrema != null && queryAnalyzer.isAggregationEnabled()) {
for (Extrema extremum : extrema) {
for (QueryField field : queryFields) {
Function<DataEvent, Object> accessor = extremum.getAccessor(field);
if (accessor != null) {
StringBuilder buf = new StringBuilder(7)
.append(channelName.getName())
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(FIELDNAME_EXTREMA)
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(extremum.name())
.append(DELIMITER_CHANNELNAME_FIELDNAME)
.append(field.name());
header.add(buf.toString());
accessors
.add(Pair.of(channelName, new QueryFieldStringifyer(accessor, EMPTY_VALUE,
DELIMITER_ARRAY)));
}
}
}
}
}
}
src/main/java/ch/psi/daq/queryrest/response/json/JSONResponseStreamWriter.java
+13 -3
View File
@@ -26,6 +26,7 @@ import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.query.model.impl.BackendQuery;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@@ -47,7 +48,8 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
private ObjectMapper mapper;
@Override
public void respond(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results, OutputStream out) throws Exception {
public void respond(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
OutputStream out) throws Exception {
AtomicReference<Exception> exception = new AtomicReference<>();
JsonGenerator generator = jsonFactory.createGenerator(out, JsonEncoding.UTF8);
@@ -78,7 +80,8 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
writer.writeValue(generator, triple.getRight());
generator.writeEndObject();
} catch (Exception e) {
LOGGER.error("Could not write channel name of channel '{}'", triple.getMiddle(), e);
LOGGER.error("Could not write channel name of channel '{}'", triple.getMiddle(),
e);
exception.compareAndSet(null, e);
}
});
@@ -106,9 +109,11 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
protected Set<String> getFields(DAQQueryElement query) {
Set<QueryField> queryFields = query.getFields();
List<Aggregation> aggregations = query.getAggregation() != null ? query.getAggregation().getAggregations() : null;
List<Extrema> extrema = query.getAggregation() != null ? query.getAggregation().getExtrema() : null;
Set<String> includedFields =
new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0)
+ (extrema != null ? extrema.size() : 0));
for (QueryField field : queryFields) {
includedFields.add(field.name());
@@ -118,6 +123,11 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
includedFields.add(aggregation.name());
}
}
if (extrema != null) {
// field of ExtremaCalculator (extrema in BinnedValueCombinedDataEvent and
// BinnedIndexCombinedDataEvent)
includedFields.add("extrema");
}
// do not write channel since it is already provided as key in mapping
includedFields.remove(QueryField.channel.name());