View Javadoc
1   package de.dlr.shepard.data.timeseries.services;
2   
3   import de.dlr.shepard.common.exceptions.InvalidAuthException;
4   import de.dlr.shepard.common.exceptions.InvalidBodyException;
5   import de.dlr.shepard.common.exceptions.InvalidPathException;
6   import de.dlr.shepard.context.references.timeseriesreference.daos.TimeseriesTupleDAO;
7   import de.dlr.shepard.data.timeseries.daos.TimeseriesDAO;
8   import de.dlr.shepard.data.timeseries.io.TimeseriesWithDataPoints;
9   import de.dlr.shepard.data.timeseries.model.Timeseries;
10  import de.dlr.shepard.data.timeseries.model.TimeseriesDataPoint;
11  import de.dlr.shepard.data.timeseries.model.TimeseriesDataPointsQueryParams;
12  import de.dlr.shepard.data.timeseries.model.TimeseriesTuple;
13  import de.dlr.shepard.data.timeseries.model.enums.DataPointValueType;
14  import de.dlr.shepard.data.timeseries.repositories.TimeseriesDataPointRepository;
15  import de.dlr.shepard.data.timeseries.utilities.ObjectTypeEvaluator;
16  import de.dlr.shepard.data.timeseries.utilities.TimeseriesValidator;
17  import io.quarkus.narayana.jta.runtime.TransactionConfiguration;
18  import jakarta.enterprise.context.RequestScoped;
19  import jakarta.enterprise.context.control.ActivateRequestContext;
20  import jakarta.inject.Inject;
21  import jakarta.transaction.Transactional;
22  import java.util.ArrayList;
23  import java.util.List;
24  import java.util.NoSuchElementException;
25  import java.util.Optional;
26  import java.util.concurrent.ConcurrentLinkedQueue;
27  import java.util.stream.Stream;
28  import org.eclipse.microprofile.config.ConfigProvider;
29  
30  @RequestScoped
31  public class TimeseriesService {
32  
33    @Inject
34    TimeseriesDAO timeseriesDAO;
35  
36    @Inject
37    TimeseriesTupleDAO timeseriesTupleDAO;
38  
39    @Inject
40    TimeseriesDataPointRepository timeseriesDataPointRepository;
41  
42    @Inject
43    TimeseriesContainerService timeseriesContainerService;
44  
45    /**
46     * Flag to determine whether integer values received should be automatically converted to double if the
47     * timeseries they are supposed to be inserted is of type double.
48     * This flag is not injected using @ConfigProperty because that would make testing much more complicated.
49     * These properties are set upon startup and cannot be changed within a single test.
50     */
51    Boolean autoConvertIntToDouble = ConfigProvider.getConfig()
52      .getOptionalValue("shepard.autoconvert-int", Boolean.class)
53      .orElse(false);
54  
55    /**
56     * Returns a list of timeseries objects that are in the given database.
57     * <p/>
58     * Returns an empty list if the timeseries container is not accessible (cannot
59     * be found or wrong permissions).
60     *
61     * @param containerId of the given timeseries container
62     * @return a list of timeseries entities
63     */
64    public Stream<Timeseries> getTimeseriesAvailable(long containerId) {
65      timeseriesContainerService.getContainer(containerId);
66      return timeseriesDAO.getAllTimeseriesInContainer(containerId);
67    }
68  
69    public Stream<Timeseries> findByQuintupleInContainer(
70      long containerId,
71      String measurement,
72      String device,
73      String location,
74      String symbolicName,
75      String field
76    ) {
77      return timeseriesDAO.findByQuintupleInContainer(containerId, measurement, device, location, symbolicName, field);
78    }
79  
80    /**
81     * Returns a timeseries entity by its timeseries id.
82     *
83     * @param id timeseries id
84     * @return timeseries
85     * @throws NoSuchElementException if the timeseries does not exist
86     * @throws InvalidAuthException   if user has no read permissions on the timeseries container
87     * @throws InvalidPathException   if container with containerId or the timeseries are not accessible
88     */
89    public Timeseries getTimeseriesById(Long id)
90      throws NoSuchElementException, InvalidAuthException, InvalidPathException {
91      var timeseries = timeseriesDAO.findByTimeseriesId(id).orElseThrow();
92      timeseriesContainerService.getContainer(timeseries.getContainer().getId());
93      return timeseries;
94    }
95  
96    /**
97     * Deletes timeseries container by id
98     *
99     * @param containerId timeseries container id
100    * @throws InvalidPathException if container could not be found
101    * @throws InvalidAuthException if user has no edit permissions on container
102    */
103   @Transactional
104   public void deleteTimeseriesByContainerId(long containerId) {
105     timeseriesContainerService.getContainer(containerId);
106     timeseriesContainerService.assertIsAllowedToDeleteContainer(containerId);
107     timeseriesDAO.deleteAllTimeseriesInContainer(containerId);
108   }
109 
110   /**
111    * Retrieve a list of DataPoints for a time-interval with options to grouping/
112    * time slicing, filling and aggregating.
113    *
114    * @return List of TimeseriesDataPoint
115    * @throws InvalidPathException if container is null or deleted
116    * @throws InvalidAuthException if user has no read permissions on container
117    */
118   public List<TimeseriesDataPoint> getDataPointsByTimeseries(
119     long containerId,
120     TimeseriesTuple timeseries,
121     TimeseriesDataPointsQueryParams queryParams
122   ) {
123     timeseriesContainerService.getContainer(containerId);
124     var ts = timeseriesDAO.findTimeseries(containerId, timeseries).orElseThrow();
125 
126     return timeseriesDataPointRepository.queryDataPoints(ts.getTimeseriesId(), ts.getValueType(), queryParams);
127   }
128 
129   @ActivateRequestContext
130   public List<TimeseriesDataPoint> getDatapointsParallelizable(
131     Timeseries timeseries,
132     TimeseriesDataPointsQueryParams queryParams
133   ) {
134     return timeseriesDataPointRepository.queryDataPoints(
135       timeseries.getTimeseriesId(),
136       timeseries.getValueType(),
137       queryParams
138     );
139   }
140 
141   public List<TimeseriesWithDataPoints> getManyTimeseriesWithDataPoints(
142     Long containerId,
143     List<TimeseriesTuple> timeseriesTupleList,
144     TimeseriesDataPointsQueryParams queryParams
145   ) {
146     timeseriesContainerService.getContainer(containerId);
147 
148     var timeseriesList = timeseriesTupleList
149       .stream()
150       .map(tsTuple -> timeseriesDAO.findTimeseries(containerId, tsTuple).orElseThrow())
151       .toList();
152 
153     ConcurrentLinkedQueue<TimeseriesWithDataPoints> timeseriesWithDataPointsQueue = new ConcurrentLinkedQueue<>();
154     timeseriesList
155       .parallelStream()
156       .forEach(timeseries ->
157         timeseriesWithDataPointsQueue.add(
158           new TimeseriesWithDataPoints(
159             timeseries.getTimeseriesTuple(),
160             getDatapointsParallelizable(timeseries, queryParams)
161           )
162         )
163       );
164     return new ArrayList<>(timeseriesWithDataPointsQueue);
165   }
166 
167   /**
168    * Saves data points in the database.
169    * If the corresponding timeseries did not exist before, it will be persisted in
170    * the database.
171    *
172    * @param timeseriesContainerId Identifies the TimeseriesContainer
173    * @param timeseries            The timeseries identifiers
174    * @param dataPoints            Data points to be added to the timeseries
175    * @return created timeseries
176    */
177   public Timeseries saveDataPoints(
178     long timeseriesContainerId,
179     TimeseriesTuple timeseries,
180     List<TimeseriesDataPoint> dataPoints
181   ) {
182     timeseriesContainerService.getContainer(timeseriesContainerId);
183     timeseriesContainerService.assertIsAllowedToEditContainer(timeseriesContainerId);
184 
185     DataPointValueType incomingValueType = ObjectTypeEvaluator.determineType(
186       dataPoints.getFirst().getValue()
187     ).orElseThrow(InvalidBodyException::new);
188 
189     return saveDataPoints(timeseriesContainerId, timeseries, dataPoints, incomingValueType);
190   }
191 
192   /**
193    * Saves data points in the database.
194    * If the corresponding timeseries did not exist before, it will be persisted in
195    * the database.
196    *
197    * @param timeseriesContainerId Identifies the TimeseriesContainer
198    * @param timeseriesTuple       The timeseries identifiers
199    * @param dataPoints            Data points to be added to the timeseries
200    * @param dataType              The data type that values in this timeseries
201    *                              will have
202    * @return created timeseries
203    */
204   @Transactional(Transactional.TxType.REQUIRES_NEW)
205   @TransactionConfiguration(timeout = 6000)
206   public Timeseries saveDataPoints(
207     long timeseriesContainerId,
208     TimeseriesTuple timeseriesTuple,
209     List<TimeseriesDataPoint> dataPoints,
210     DataPointValueType dataType
211   ) {
212     var ts = getTimeseries(timeseriesContainerId, timeseriesTuple).orElseGet(() ->
213       createTimeseries(timeseriesContainerId, timeseriesTuple, dataType)
214     );
215     assertDataPointsMatchTimeseriesValueType(ts.getValueType(), dataPoints);
216     timeseriesDataPointRepository.insertManyDataPoints(dataPoints, ts.getTimeseriesId(), ts.getValueType());
217     return ts;
218   }
219 
220   public Optional<Timeseries> getTimeseries(long containerId, TimeseriesTuple timeseries) {
221     return timeseriesDAO.findTimeseries(containerId, timeseries);
222   }
223 
224   /**
225    * Persist a {@link Timeseries} in Neo4j.
226    * If its referenced {@link TimeseriesTuple} does not exist yet create it.
227    *
228    *
229    * @param containerId The ID of the container that the timeseries is associated with
230    * @param timeseriesTuple The {@link TimeseriesTuple} referenced by the timeseries.
231    *                        If it exists the timeseries will reference the existing tuple.
232    *                        If it does not exist yet, it is newly created.
233    * @param incomingValueType The value type the timeseries should have
234    * @return The created {@link Timeseries} as found in the database.
235    */
236   private synchronized Timeseries createTimeseries(
237     long containerId,
238     TimeseriesTuple timeseriesTuple,
239     DataPointValueType incomingValueType
240   ) {
241     timeseriesContainerService.assertIsAllowedToEditContainer(containerId);
242     TimeseriesValidator.assertTimeseriesPropertiesAreValid(timeseriesTuple);
243     var container = timeseriesContainerService.getContainer(containerId);
244     // If TimeseriesTuple already exists update it, mainly with the id so the framework can handle the update correctly
245     timeseriesTuple = timeseriesTupleDAO.find(timeseriesTuple).orElse(timeseriesTuple);
246     var tsToCreate = new Timeseries(
247       container,
248       timeseriesTuple,
249       incomingValueType,
250       timeseriesDAO.getCurrentMaximumTimeseriesId() + 1
251     );
252     return timeseriesDAO.createOrUpdate(tsToCreate);
253   }
254 
255   private void assertDataPointsMatchTimeseriesValueType(
256     DataPointValueType valueType,
257     List<TimeseriesDataPoint> dataPoints
258   ) {
259     for (TimeseriesDataPoint dataPoint : dataPoints) {
260       DataPointValueType expectedType = ObjectTypeEvaluator.determineType(dataPoint.getValue()).orElseThrow(
261         InvalidBodyException::new
262       );
263       assertValueTypeMatchesTimeseries(valueType, expectedType);
264     }
265   }
266 
267   private void assertValueTypeMatchesTimeseries(DataPointValueType tsValueType, DataPointValueType incomingValueType) {
268     // If auto-conversion is enabled, allow transformation from Integer to Double
269     if (
270       autoConvertIntToDouble &&
271       incomingValueType == DataPointValueType.Integer &&
272       tsValueType == DataPointValueType.Double
273     ) return;
274 
275     if (tsValueType != incomingValueType) throw new InvalidBodyException(
276       "Timeseries already exists for data type %s but new data points are of type %s",
277       tsValueType,
278       incomingValueType
279     );
280   }
281 }