def get_points(self, query, count=10, scope='*'):
# type: (str, int, str) -> osisoftpy.structures.TypedList[Point]
"""
:param query:
:param count:
:param scope:
:return:
"""
payload = {'q': query, 'count': count, 'scope': scope}
log.debug(
'Executing Query against PI Web WebAPI Indexed Search with '
'the following parameters: Query: "%s", Count: "%s". Payload: %s',
query, count, payload)
r = self.session.get(self.url + '/search/query', params=payload)
if r.status_code != requests.codes.ok:
r.raise_for_status()
else:
data = r.json()
log.debug('HTTP %s - Instantiating %s PI points', r.status_code,
get_count(data.get('Items', None)))
factory = Factory(Point)
items = list(map(lambda x: create(factory, x),
data.get('Items', None)))
points = TypedList(Point)
for point in items:
points.append(point)
log.debug('PI Point retrieval success! %s PI '
'point(s) were '
'found and instantiated.', get_count(points))
if len(data['Errors']) != 0:
for error in data['Errors']:
try:
log.warning('The PI Web WebAPI returned the '
'following error while instantiating '
'PI points. '
'ErrorCode: {0}, Source: {1}, '
'Message {2}'.format(
error['ErrorCode'], error['Source'],
error['Message']))
except Exception as e:
log.error('Exception encounted while '
'instantiating '
'PI points!', exc_info=True)
return points
评论列表
文章目录