diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 05c655a180..21d5787cc6 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -39,17 +39,17 @@ import org.openremote.model.attribute.AttributeRef; import org.openremote.model.datapoint.AssetDatapointResource; import org.openremote.model.datapoint.DatapointPeriod; +import org.openremote.model.datapoint.DatapointExportFormat; import org.openremote.model.datapoint.DatapointQueryTooLargeException; import org.openremote.model.datapoint.ValueDatapoint; import org.openremote.model.datapoint.query.AssetDatapointQuery; import org.openremote.model.http.RequestParams; import org.openremote.model.security.ClientRole; import org.openremote.model.syslog.SyslogCategory; +import org.openremote.model.util.UniqueIdentifierGenerator; import org.openremote.model.value.MetaItemType; -import java.io.File; -import java.io.FileInputStream; -import java.util.concurrent.ScheduledFuture; +import java.io.*; import java.util.logging.Level; import java.util.logging.Logger; import java.util.zip.ZipEntry; @@ -174,10 +174,18 @@ public DatapointPeriod getDatapointPeriod(RequestParams requestParams, String as } @Override - public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp) { + public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, DatapointExportFormat format) { try { + if (format == null) { + throw new WebApplicationException(Response.Status.BAD_REQUEST); + } + AttributeRef[] attributeRefs = JSON.readValue(attributeRefsString, AttributeRef[].class); + if (attributeRefs == null || attributeRefs.length == 0) { + throw new WebApplicationException(Response.Status.BAD_REQUEST); + } + for (AttributeRef attributeRef : attributeRefs) { if (isRestrictedUser() && !assetStorageService.isUserAsset(getUserId(), attributeRef.getId())) { throw new WebApplicationException(Response.Status.FORBIDDEN); @@ -199,49 +207,45 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs ); } - DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp); + DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp + " in format " + format); - ScheduledFuture exportFuture = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp); + PipedInputStream pipedInputStream = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); - asyncResponse.register((ConnectionCallback) disconnected -> exportFuture.cancel(true)); - - File exportFile = null; + asyncResponse.register((ConnectionCallback) disconnected -> { + try { + pipedInputStream.close(); + } catch (IOException e) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Could not close input stream: ", e); + throw new RuntimeException(e); + } + }); - try { - exportFile = exportFuture.get(); + try (InputStream fin = pipedInputStream; + ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream())) { - ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream()); - FileInputStream fin = new FileInputStream(exportFile); - ZipEntry zipEntry = new ZipEntry(exportFile.getName()); + String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; + ZipEntry zipEntry = new ZipEntry(fileName); zipOut.putNextEntry(zipEntry); IOUtils.copy(fin, zipOut); zipOut.closeEntry(); - zipOut.close(); - fin.close(); - - response.setContentType("application/zip"); - response.setHeader(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"dataexport.zip\""); - - asyncResponse.resume( - response - ); - } catch (Exception ex) { - exportFuture.cancel(true); + } catch (IOException ex) { asyncResponse.resume(new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR)); - DATA_EXPORT_LOG.log(Level.SEVERE, "Exception in ScheduledFuture: ", ex); - } finally { - if (exportFile != null && exportFile.exists()) { - try { - exportFile.delete(); - } catch (Exception e) { - DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to delete temporary export file: " + exportFile.getPath(), e); - } - } + DATA_EXPORT_LOG.log(Level.SEVERE, "Zip exception: ", ex); } + + response.setContentType("application/zip"); + response.setHeader(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"dataexport.zip\""); + + asyncResponse.resume( + response + ); } catch (JsonProcessingException ex) { asyncResponse.resume(new BadRequestException(ex)); } catch (DatapointQueryTooLargeException dqex) { asyncResponse.resume(new WebApplicationException(dqex, Response.Status.REQUEST_ENTITY_TOO_LARGE)); + } catch (IOException ex) { + asyncResponse.resume(new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR)); + DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to create piped output stream: ", ex); } } } diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 050ec94358..21572721b8 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -4,8 +4,8 @@ import org.openremote.agent.protocol.ProtocolDatapointService; import org.openremote.container.timer.TimerService; import org.openremote.manager.asset.OutdatedAttributeEvent; +import org.openremote.model.datapoint.DatapointExportFormat; import org.openremote.model.datapoint.DatapointQueryTooLargeException; -import org.openremote.model.util.UniqueIdentifierGenerator; import org.openremote.manager.asset.AssetProcessingException; import org.openremote.manager.asset.AssetStorageService; import org.openremote.manager.event.ClientEventService; @@ -21,13 +21,14 @@ import org.openremote.model.query.AssetQuery; import org.openremote.model.query.filter.AttributePredicate; import org.openremote.model.query.filter.NameValuePredicate; +import org.openremote.model.syslog.SyslogCategory; import org.openremote.model.util.Pair; import org.openremote.model.value.MetaHolder; import org.openremote.model.value.MetaItemType; +import org.postgresql.PGConnection; +import org.postgresql.copy.CopyManager; -import java.io.File; -import java.nio.file.Files; -import java.nio.file.Path; +import java.io.*; import java.sql.Date; import java.sql.PreparedStatement; import java.sql.Statement; @@ -36,15 +37,17 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; import java.util.stream.Collectors; +import java.util.Set; +import java.util.TreeSet; import static java.time.temporal.ChronoUnit.DAYS; import static java.util.stream.Collectors.groupingBy; import static java.util.stream.Collectors.toList; +import static org.openremote.model.syslog.SyslogCategory.DATA; import static org.openremote.model.util.MapAccess.getInteger; import static org.openremote.model.value.MetaItemType.STORE_DATA_POINTS; @@ -61,10 +64,9 @@ public class AssetDatapointService extends AbstractDatapointService - file.isFile() - && file.getName().endsWith("csv") - && file.lastModified() < timerService.getCurrentTimeMillis() - oneDayMillis - ); - - if (obsoleteExports != null) { - Arrays.stream(obsoleteExports) - .forEach(file -> { - boolean success = false; - try { - success = file.delete(); - } catch (SecurityException e) { - LOG.log(Level.WARNING, "Cannot access the export file to delete it", e); - } - if (!success) { - LOG.log(Level.WARNING, "Failed to delete obsolete export '" + file.getName() + "'"); - } - }); - } - } catch (Exception e) { - LOG.log(Level.WARNING, "Failed to purge old exports", e); - } } protected String buildWhereClause(List>> attributes, boolean negate) { @@ -252,16 +218,28 @@ protected String buildWhereClause(List>> attributes, b /** * Exports datapoints as CSV using SQL; the export path used in the SQL query must also be mapped into the manager * container so it can be accessed by this process. + * Backwards compatible overload with default format. */ - public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, + public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, - long toTimestamp) { + long toTimestamp) throws IOException { + return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, DatapointExportFormat.CSV); + } + + /** + * Exports datapoints as CSV using SQL; the export path used in the SQL query must also be mapped into the manager + * container so it can be accessed by this process. + */ + public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, + long fromTimestamp, + long toTimestamp, + DatapointExportFormat format) throws IOException { try { ExportQuery exportQuery = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); // Verify the query is 'legal' and can be executed if(canQueryDatapoints(exportQuery.query, exportQuery.parameters, datapointExportLimit)) { - return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp); + return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); } throw new RuntimeException("Could not export datapoints."); @@ -272,43 +250,156 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, } } - protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, + protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, - long toTimestamp) { - - return scheduledExecutorService.schedule(() -> { - String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - String tempTableName = "tmp_export_attributes"; - String copySql = buildCopyExportSql(tempTableName, fileName, fromTimestamp, toTimestamp); - - persistenceService.doTransaction(em -> em.unwrap(Session.class).doWork(connection -> { - // It is not possible to use parametrized prepared statements with the copy statement, - // so we're going through a temporary table for safely binding the parameters - try (Statement statement = connection.createStatement()) { - statement.execute( - "create temp table " + tempTableName + " (entity_id text, attribute_name text) on commit drop" - ); - } + long toTimestamp, + DatapointExportFormat format) throws IOException { + // Increase buffer size (default is only 1 KB) + PipedInputStream in = new PipedInputStream(1024 * 1024 * 4); // 4 MB + PipedOutputStream out = new PipedOutputStream(in); + + // Execute query asynchronously + scheduledExecutorService.schedule(() -> { + boolean success = false; + try { + persistenceService.doTransaction(em -> em.unwrap(Session.class).doWork(connection -> { + String tempTableName = "tmp_export_attributes"; + + // Create temp table and insert attribute refs safely via PreparedStatement + // to prevent SQL injection (instead of string-interpolating IDs into the query) + try (Statement statement = connection.createStatement()) { + statement.execute( + "create temp table " + tempTableName + " (entity_id text, attribute_name text) on commit drop" + ); + } - try (PreparedStatement insertStatement = connection.prepareStatement( - "insert into " + tempTableName + " (entity_id, attribute_name) values (?, ?)" - )) { - for (AttributeRef attributeRef : attributeRefs) { - insertStatement.setString(1, validateAssetId(attributeRef.getId())); - insertStatement.setString(2, attributeRef.getName()); - insertStatement.addBatch(); + try (PreparedStatement insertStatement = connection.prepareStatement( + "insert into " + tempTableName + " (entity_id, attribute_name) values (?, ?)" + )) { + for (AttributeRef attributeRef : attributeRefs) { + insertStatement.setString(1, validateAssetId(attributeRef.getId())); + insertStatement.setString(2, attributeRef.getName()); + insertStatement.addBatch(); + } + insertStatement.executeBatch(); } - insertStatement.executeBatch(); - } - try (Statement statement = connection.createStatement()) { - statement.execute(copySql); - } - })); + // Build the COPY ... TO STDOUT query based on format + String copyQuery = buildCopyToStdoutQuery(tempTableName, fromTimestamp, toTimestamp, format, attributeRefs); - // The same path must resolve in both the postgresql container and the manager container - return exportPath.resolve(fileName).toFile(); + try { + PGConnection pgConnection = connection.unwrap(PGConnection.class); + CopyManager copyManager = pgConnection.getCopyAPI(); + copyManager.copyOut(copyQuery, out); + out.flush(); + } catch (IOException e) { + throw new RuntimeException("Failed to export datapoints", e); + } + })); + success = true; + } catch (Exception e) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Datapoint export failed", e); + } finally { + try { + out.close(); + } catch (IOException e) { + DATA_EXPORT_LOG.log(Level.WARNING, "Failed to close output stream", e); + } + if (!success) { + try { + in.close(); + } catch (IOException e) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to close piped input stream", e); + } + } + } }, 0, TimeUnit.MILLISECONDS); + + return in; + } + + /** + * Builds a COPY ... TO STDOUT query for streaming CSV data based on the export format. + * Uses a temporary table for safe attribute filtering (SQL injection prevention). + */ + private String buildCopyToStdoutQuery(String tempTableName, long fromTimestamp, long toTimestamp, + DatapointExportFormat format, AttributeRef[] attributeRefs) { + long fromSeconds = fromTimestamp / 1000; + long toSeconds = toTimestamp / 1000; + final String TO_STDOUT_WITH_CSV_FORMAT = ") TO STDOUT WITH (FORMAT CSV, HEADER, DELIMITER ',');"; + + if (format == DatapointExportFormat.CSV_CROSSTAB || format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + Set headers = getAttributeHeaders(attributeRefs); + + // Category query using VALUES clause with dollar-quoting to avoid escaping issues + String categoryValues = headers.stream() + .map(header -> "('" + header.replace("'", "''") + "')") + .collect(Collectors.joining(", ")); + String categoryQuery = "SELECT header FROM (VALUES " + categoryValues + ") AS t(header)"; + + // Column definitions for crosstab result + String attributeColumns = headers.stream() + .map(header -> "\"" + header + "\" text") + .collect(Collectors.joining(", ")); + + // Inner SELECT query for crosstab (uses escaped quotes since it's inside a string argument) + String innerQuery; + if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + innerQuery = String.format( + "select public.time_bucket(''1 minute'', ad.timestamp) as bucket_timestamp, " + + "a.name || '' : '' || ad.attribute_name as header, " + + "CASE " + + " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + + " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + + " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + + "END as value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "group by bucket_timestamp, header " + + "order by bucket_timestamp, header", + tempTableName, fromSeconds, toSeconds); + } else { + innerQuery = String.format( + "select ad.timestamp, a.name || '' : '' || ad.attribute_name as header, ad.value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "order by ad.timestamp, header", + tempTableName, fromSeconds, toSeconds); + } + + return String.format( + "copy (select * from crosstab('%s', $cat$%s$cat$) as ct(timestamp timestamp, %s)%s", + innerQuery, categoryQuery, attributeColumns, TO_STDOUT_WITH_CSV_FORMAT); + } else { + // Plain CSV format using temp table join + String innerQuery = String.format( + "select ad.timestamp, a.name, ad.attribute_name, value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "order by ad.timestamp desc", + tempTableName, fromSeconds, toSeconds); + return "copy (" + innerQuery + TO_STDOUT_WITH_CSV_FORMAT; + } + } + + /** + * Function for building CSV attribute headers when format will make + * a separate column per attribute. + * Uses TreeSet for alphabetical ordering required by crosstab queries. + */ + private Set getAttributeHeaders(AttributeRef[] attributeRefs) { + Set headers = new TreeSet<>(); + Arrays.stream(attributeRefs).forEach(attr -> { + String assetName = assetStorageService.findNames(attr.getId()).toString().replaceAll("(^\\[|\\]$)", ""); + headers.add(assetName + " : " + attr.getName()); + }); + return headers; } /** @@ -343,27 +434,6 @@ protected ExportQuery getSelectExportQuery(AttributeRef[] attributeRefs, long fr protected record ExportQuery(String query, Map parameters) { } - private static String buildCopyExportSql(String tempTableName, String fileName, long fromTimestamp, long toTimestamp) { - long fromSeconds = fromTimestamp / 1000; - long toSeconds = toTimestamp / 1000; - - StringBuilder sb = new StringBuilder("copy (") - .append("select ad.timestamp, a.name, ad.attribute_name, value ") - .append("from asset_datapoint ad ") - .append("join asset a on ad.entity_id = a.id ") - .append("join ").append(tempTableName).append(" t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name ") - .append("where ad.timestamp >= to_timestamp(").append(fromSeconds).append(") ") - .append("and ad.timestamp <= to_timestamp(").append(toSeconds).append(") ") - .append("order by ad.timestamp desc") - .append(") to '/storage/") - .append(EXPORT_STORAGE_DIR_NAME) - .append("/") - .append(fileName) - .append("' delimiter ',' CSV HEADER;"); - - return sb.toString(); - } - private static String validateAssetId(String assetId) { if (assetId == null || assetId.length() != 22) { throw new IllegalArgumentException("Invalid asset id"); diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20260120_01__addTableFunc.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20260120_01__addTableFunc.sql new file mode 100644 index 0000000000..c24323fa87 --- /dev/null +++ b/manager/src/main/resources/org/openremote/manager/setup/database/V20260120_01__addTableFunc.sql @@ -0,0 +1 @@ +CREATE EXTENSION IF NOT EXISTS tablefunc; diff --git a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java index d2a182dc16..d98e919a28 100644 --- a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java +++ b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java @@ -69,6 +69,7 @@ DatapointPeriod getDatapointPeriod(@BeanParam RequestParams requestParams, void getDatapointExport(@Suspended AsyncResponse asyncResponse, @QueryParam("attributeRefs") String attributeRefsString, @QueryParam("fromTimestamp") long fromTimestamp, - @QueryParam("toTimestamp") long toTimestamp); + @QueryParam("toTimestamp") long toTimestamp, + @QueryParam("format") @DefaultValue("CSV") DatapointExportFormat format); } diff --git a/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java b/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java new file mode 100644 index 0000000000..4ae4e72479 --- /dev/null +++ b/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java @@ -0,0 +1,27 @@ +/* + * Copyright 2025, OpenRemote Inc. + * + * See the CONTRIBUTORS.txt file in the distribution for a + * full listing of individual contributors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package org.openremote.model.datapoint; + +public enum DatapointExportFormat { + + CSV, + CSV_CROSSTAB, + CSV_CROSSTAB_MINUTE +} diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index 9e86864d59..046b172c8a 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -6,6 +6,7 @@ import org.openremote.manager.setup.SetupService import org.openremote.manager.web.ManagerWebService import org.openremote.model.asset.impl.LightAsset import org.openremote.model.attribute.AttributeRef +import org.openremote.model.datapoint.DatapointExportFormat import org.openremote.model.datapoint.DatapointQueryTooLargeException import org.openremote.model.datapoint.ValueDatapoint import org.openremote.model.query.AssetQuery @@ -71,18 +72,15 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer ] ) - then: "the CSV export should return a file" - def csvExport1Future = assetDatapointService.exportDatapoints( + then: "the default CSV export should return a file" + def inputStream1 = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() ) - assert csvExport1Future != null - def csvExport1 = csvExport1Future.get() - assert csvExport1 != null and: "the CSV export should contain all 5 data points" - def csvExport1Lines = csvExport1.readLines() + def csvExport1Lines = inputStream1.readLines() assert csvExport1Lines.size() == 6 assert csvExport1Lines[1].endsWith("10.0") assert csvExport1Lines[2].endsWith("20.0") @@ -92,11 +90,62 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ + when: "exporting with format CSV_CROSSTAB" + def inputStream2 = assetDatapointService.exportDatapoints( + [new AttributeRef(asset.id, attributeName)] as AttributeRef[], + dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + DatapointExportFormat.CSV_CROSSTAB + ) + + then: "the crosstab CSV export should contain all 5 data points in columnar format" + def csvExport2Lines = inputStream2.readLines() + assert csvExport2Lines.size() == 6 // header + 5 data rows + assert csvExport2Lines[0].contains("timestamp") + assert csvExport2Lines[0].contains(assetName + " : " + attributeName) + + /* ------------------------- */ + + when: "exporting with format CSV_CROSSTAB_MINUTE" + def inputStream3 = assetDatapointService.exportDatapoints( + [new AttributeRef(asset.id, attributeName)] as AttributeRef[], + dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + DatapointExportFormat.CSV_CROSSTAB_MINUTE + ) + + then: "the time-bucketed CSV export should contain aggregated data with correct values" + def csvExport3Lines = inputStream3.readLines() + assert csvExport3Lines.size() >= 2 // header + at least 1 aggregated row + assert csvExport3Lines[0].contains("timestamp") + assert csvExport3Lines[0].contains(assetName + " : " + attributeName) + + and: "the aggregated values should be correct averages within 1-minute buckets" + // Each datapoint is 5 minutes apart, so each should be in its own 1-minute bucket + // Therefore we should have 5 rows of data (one per bucket) + assert csvExport3Lines.size() == 6 // header + 5 buckets + + // Extract the values from the CSV (skip header, get last column which is the brightness value) + def values = csvExport3Lines[1..5].collect { line -> + def parts = line.split(',') + parts.size() > 1 ? parts[1].trim() : null + }.findAll { it != null && it != '' } + + // Should have 5 values, one per bucket, matching our input: 10, 20, 30, 40, 50 + assert values.size() == 5 + assert values.contains("10.000") || values.contains("10") + assert values.contains("20.000") || values.contains("20") + assert values.contains("30.000") || values.contains("30") + assert values.contains("40.000") || values.contains("40") + assert values.contains("50.000") || values.contains("50") + + /* ------------------------- */ + when: "the limit of data export is lowered to 4" assetDatapointService.datapointExportLimit = 4 and: "we try to export the same amount of data points" - def csvExport2Future = assetDatapointService.exportDatapoints( + def inputStream4 = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() @@ -138,14 +187,13 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer when: "exporting with a malicious attribute name" def injectedAttributeName = "x')) UNION ((SELECT now()::timestamp, table_name::text, 'injected'::text, NULL::jsonb FROM information_schema.tables WHERE table_schema='public" - def csvExportFuture = assetDatapointService.exportDatapoints( + def csvExport = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, injectedAttributeName)] as AttributeRef[], LocalDateTime.now().minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() ) then: "the CSV export should not contain injected rows" - def csvExport = csvExportFuture.get() assert csvExport != null def csvExportLines = csvExport.readLines() assert csvExportLines.size() == 1 diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 1398d2cf96..92e7612d3e 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -8,7 +8,7 @@ import {i18next} from "@openremote/or-translate"; import manager, { DefaultColor3, Util } from "@openremote/core"; import { InputType, OrInputChangedEvent } from "@openremote/or-mwc-components/or-mwc-input"; import {OrAssetAttributePickerPickedEvent, OrAssetAttributePicker} from "@openremote/or-attribute-picker"; -import { AttributeRef } from "@openremote/model"; +import { AttributeRef, DatapointExportFormat } from "@openremote/model"; import moment from "moment"; import { buttonStyle } from "@openremote/or-rules"; import {createSelector} from "reselect"; @@ -16,6 +16,7 @@ import {isAxiosError} from "@openremote/rest"; import { showSnackbar } from "@openremote/or-mwc-components/or-mwc-snackbar"; import { showDialog } from "@openremote/or-mwc-components/or-mwc-dialog"; const tableStyle = require("@material/data-table/dist/mdc.data-table.css"); +import { when } from "lit/directives/when.js"; export function pageExportProvider(store: Store): PageProvider { return { @@ -175,9 +176,12 @@ export class PageExport extends Page { } .timerange-wrapper { + margin-bottom: 1em + } + .options { + width: fit-content; margin-bottom: 2em } - .export-btn-wrapper { display: flex; justify-content: space-between; @@ -200,6 +204,18 @@ export class PageExport extends Page { @property({type: Number}) private latestTimestamp: number = moment().valueOf(); + @property() + protected _loading: boolean = false; + + protected _exportFormats: Record = { + [DatapointExportFormat.CSV]: i18next.t('exportFormatCSV'), + [DatapointExportFormat.CSV_CROSSTAB]: i18next.t('exportFormatCSVCrosstab'), + [DatapointExportFormat.CSV_CROSSTAB_MINUTE]: i18next.t('exportFormatCSVCrosstabMinute') + }; + + @property() + private selectedFormat: DatapointExportFormat = DatapointExportFormat.CSV; + private config?: OrExportConfig; private realm: string; private isClearExportBtnDisabled: boolean = true; @@ -240,11 +256,17 @@ export class PageExport extends Page { ${i18next.t("dataExport")} +

${i18next.t("dataSelection")}

+ ${when(this._loading, () => html` +
+ +
+ `)} ${headers.map(header => html` @@ -262,9 +284,14 @@ export class PageExport extends Page {
${header}
-
- - +
+
+ + +
+
+ +
@@ -290,7 +317,7 @@ export class PageExport extends Page { attributeName: attrRef.name, }); }); - + this._loading = true; Promise.all(dataPointInfoPromises).then(datapointPeriod => { const assetInfoPromises = datapointPeriod.map(result => { @@ -301,6 +328,7 @@ export class PageExport extends Page { const allAssets = assetInfos.map(attr => attr.data); const allDatapoints = datapointPeriod.map(datapoints => datapoints.data); + if (allDatapoints.length > 0) { this.tableRows = allDatapoints.map(dataInfo => { const relevantAssetInfo = allAssets.find(asset => asset.id === dataInfo.assetId); @@ -315,6 +343,7 @@ export class PageExport extends Page { this.renderTableRows(); this.isClearExportBtnDisabled = false; this.isExportBtnDisabled = false; + this._loading = false; } else { this.isClearExportBtnDisabled = true; this.isExportBtnDisabled = true; @@ -375,10 +404,12 @@ export class PageExport extends Page { } protected export = () => { + this._loading = true; manager.rest.api.AssetDatapointResource.getDatapointExport({ attributeRefs: JSON.stringify(this.tableRows.map(attr => ({id: attr.assetId, name: attr.attributeName}))), fromTimestamp: this.oldestTimestamp, - toTimestamp: this.latestTimestamp + toTimestamp: this.latestTimestamp, + format: this.selectedFormat }, { responseType: "blob", @@ -391,13 +422,20 @@ export class PageExport extends Page { link.setAttribute("download", "dataexport.zip"); document.body.appendChild(link); link.click(); - + this._loading = false; }).catch(ex => { if(isAxiosError(ex)) { if(ex.response?.status === 413) { showSnackbar(undefined, "exportTooLargeError"); } + else if(ex.message.includes("timeout")) { + showSnackbar(undefined, "noAttributeDataTimeout"); + } + else { + showSnackbar(undefined, "errorOccurred"); + } } + this._loading = false; }); } diff --git a/ui/app/manager/test/export.test.ts b/ui/app/manager/test/export.test.ts new file mode 100644 index 0000000000..4fb88e7224 --- /dev/null +++ b/ui/app/manager/test/export.test.ts @@ -0,0 +1,112 @@ +import { expect } from "@openremote/test"; +import { test, userStatePath } from "./fixtures/manager"; +import { preparedAssetsForRules } from "./fixtures/data/assets"; +import { AttributeEvent } from "@openremote/model"; + +test.use({ storageState: userStatePath }); + +test.beforeEach(async ({ manager }) => { + await manager.setup("smartcity", { assets: exportAssets }); +}); + +test.afterEach(async ({ manager }) => { + await manager.cleanUp(); +}); + +// Reuse assets from rules test which have storeDataPoints: true +const exportAssets = preparedAssetsForRules; +const exportAttributeName = "energyLevel"; + +/** + * Generate attribute events with timestamps for creating historical datapoints. + * @param assetId The asset ID to create events for + * @param attributeName The attribute name to create events for + * @param count Number of datapoints to create + * @param intervalMinutes Interval between datapoints in minutes + */ +function generateDatapointEvents(assetId: string, attributeName: string, count: number = 5, intervalMinutes: number = 5): AttributeEvent[] { + const now = Date.now(); + const events: AttributeEvent[] = []; + + for (let i = 0; i < count; i++) { + events.push({ + eventType: "attribute", + ref: { + id: assetId, + name: attributeName + }, + value: 20 + i * 2, + timestamp: now - (count - 1 - i) * intervalMinutes * 60 * 1000 + }); + } + + return events; +} + +/** + * Add an attribute to the export table using the attribute picker dialog. + * @param page The page to find "add asset attribute" on + * @param assetName The name of the asset to select in the asset tree + * @param attributeName The name of the attribute to select + */ +async function addAttributeViaPicker(page: import("@playwright/test").Page, assetName: string, attributeName: string) { + await page.locator("a.button:has-text('Add asset attribute')").click(); + const dialog = page.locator("or-asset-attribute-picker"); + await dialog.locator(".mdc-dialog--open").waitFor(); + + const assetTree = dialog.locator("or-asset-tree"); + await assetTree.locator("li.asset-list-element").first().waitFor(); + await assetTree.locator(`li.asset-list-element:has-text("${assetName}")`).click(); + + const attributeList = dialog.locator("#attribute-selector"); + await attributeList.waitFor(); + const displayName = attributeName.replace(/([a-z])([A-Z])/g, "$1 $2").replace(/^./, c => c.toUpperCase()); + const attributeCheckbox = attributeList.locator("li.mdc-list-item").filter({ hasText: displayName }); + await attributeCheckbox.click(); + + await dialog.locator("#add-btn").click(); + await dialog.locator(".mdc-dialog--open").waitFor({ state: "detached" }); +} + +/** + * @given An asset with datapoints exists + * @and Logged into the "smartcity" realm + * @and Navigated to the "Export" page + * @when Selecting an attribute with datapoints + * @and Clicking export + * @then The API call to the file export download should return ok + */ +test("Export datapoints successfully triggers download", async ({ page, manager }) => { + const asset = manager.assets[0]; + const events = generateDatapointEvents(asset.id!, exportAttributeName, 5); + await manager.api.AssetResource.writeAttributeEvents(events); + + await manager.goToRealmStartPage("smartcity"); + await page.waitForLoadState("networkidle"); + await manager.navigateToMenuItem("export"); + await page.locator("text=Data export").waitFor({ state: "visible" }); + + await addAttributeViaPicker(page, asset.name!, exportAttributeName); + await page.locator("table tbody tr").first().waitFor({ state: "visible", timeout: 15000 }); + + const oldestDatapoint = await page.locator("table tbody tr").first().locator("td").nth(2).textContent(); + const latestDatapoint = await page.locator("table tbody tr").first().locator("td").nth(3).textContent(); + const toDatetimeLocal = (dateStr: string) => { + const date = new Date(dateStr); + return date.toISOString().slice(0, 16); + }; + await page.locator("or-mwc-input[label='Export from']").locator("input").fill(toDatetimeLocal(oldestDatapoint!)); + await page.locator("or-mwc-input[label='To']").locator("input").fill(toDatetimeLocal(latestDatapoint!)); + + await page.locator("or-mwc-input[label='Export format']").click(); + await page.locator("li.mdc-list-item:has-text('CSV')").first().click(); + + const exportResponsePromise = page.waitForResponse( + response => response.url().includes("/asset/datapoint/export") + ); + await page.getByRole("button", { name: "Export", exact: true }).click(); + + const response = await exportResponsePromise; + expect(response.status()).toBe(200); + await expect(page.locator("or-mwc-snackbar .mdc-snackbar--open")).not.toBeAttached({ timeout: 2000 }); +}); diff --git a/ui/app/shared/locales/en/or.json b/ui/app/shared/locales/en/or.json index 4e233950f2..8d8f2db8dd 100644 --- a/ui/app/shared/locales/en/or.json +++ b/ui/app/shared/locales/en/or.json @@ -693,6 +693,10 @@ "writePermissions": "Write permissions", "valueType": "Value type", "dataExport": "Data export", + "exportFormat": "Export format", + "exportFormatCSV": "CSV", + "exportFormatCSVCrosstab": "CSV crosstab", + "exportFormatCSVCrosstabMinute": "CSV crosstab (1 min interval)", "dataSelection": "Data selection", "exportFrom": "Export from", "assetName": "Asset name", diff --git a/ui/app/shared/locales/nl/or.json b/ui/app/shared/locales/nl/or.json index 76df3e57dd..5b805878e0 100644 --- a/ui/app/shared/locales/nl/or.json +++ b/ui/app/shared/locales/nl/or.json @@ -689,6 +689,10 @@ "writePermissions": "Schrijfrechten", "valueType": "Waarde type", "dataExport": "Data export", + "exportFormat": "Export format", + "exportFormatCSV": "CSV", + "exportFormatCSVCrosstab": "CSV kruistabel", + "exportFormatCSVCrosstabMinute": "CSV kruistabel (1 min interval)", "dataSelection": "Data selectie", "exportFrom": "Exporteer vanaf", "assetName": "Asset naam",