Extracting recorded values¶
The data in the PI archives are typically compressed [1]. To get the exact values
as they are stored in the archive, the recorded_values method should be
used. It is also possible to extract a single historic value using recorded_value.
This is available on both PIPoint
, and PIAFAttribute
objects.
For simplicity this tutorial only uses PIPoint
objects,
see the tutorial on PI AF to find how to access
PIAFAttribute
objects.
Single vs Multiple values¶
We start of by extracting a the value from the first PIPoint
that is returned by the server as it was 5 minutes ago.
import PIconnect as PI
with PI.PIServer() as server:
point = server.search('*')[0]
data = point.recorded_value('-5m')
print(data)
You will see PISeries
is printed containing a single row, with the PIPoint name
as the Series name, the point value as the value, and the corresponding timestamp as
the index.
By default the PI Server automatically detects which value to select and returns it with the requested timestamp as the index. To control which value is returned, we pass the retrieval_mode argument to recorded_value:
import PIconnect as PI
from PIconnect.PIConsts import RetrievalMode
with PI.PIServer() as server:
point = server.search('*')[0]
data = point.recorded_value('-5m', retrieval_mode=RetrievalMode.AT_OR_BEFORE)
print(data)
Since it is unlikely there is a value at exactly 5 minutes ago, the PI Server now returns the latest value that’s before the requested time. Also the index is now no longer at the requested time, but at the time the value was recorded.
Now to get a time series of the values from the PIPoint
we use
the recorded_values method, and pass a start_time and end_time:
import PIconnect as PI
with PI.PIServer() as server:
points = server.search('*')[0]
data = points.recorded_values('*-48h', '*')
print(data)
Boundary types¶
By default only the data between the start_time and end_time is returned. It is also possible to instead return the data from the last value before start_time up to and including the first value after end_time, by setting the boundary_type to outside:
import PIconnect as PI
with PI.PIServer() as server:
points = server.search('*')[0]
data = points.recorded_values('*-48h', '*', boundary_type='outside')
print(data)
Warning
The boundary_type
argument currently takes a string as
the key to the internal __boundary_types
dictionary.
This will change in a future version to an enumeration in
PIConsts
.
Finally, it is also possible to interpolate the values surrounding both boundaries such that a value is returned exactly at the requested timestamp:
import PIconnect as PI
with PI.PIServer() as server:
points = server.search('*')[0]
data = points.recorded_values('*-48h', '*', boundary_type='interpolate')
print(data)
Filtering values¶
Sometimes it is desirable to exclude certain values from the returned data.
This is possible using the filter_expression argument of the
PIPoint.recorded_values
method. Only values matching the expression are
returned.
The simplest test is to only return values below a given value. To test if the
values of a tag called Plant1_Flow_out are below the value 100, you need the
filter_expression=“‘Plant1_Flow_out’ < 100”. PIPoint.recorded_values
provides a shortcut to include the tag name, by replacing %tag% with the
current tag name:
import PIconnect as PI
with PI.PIServer() as server:
points = server.search('*')[0]
print(points.recorded_values(
'*-48h',
'*',
filter_expression="'%tag%' < 115"
))
Multiple tests can be combined with the keywords and and or:
import PIconnect as PI
with PI.PIServer() as server:
points = server.search('*')[0]
print(points.recorded_values(
'*-48h',
'*',
filter_expression="'%tag%' > 100 and '%tag%' < 115"
))