mirror of
https://github.com/Unidata/python-awips.git
synced 2025-02-24 06:57:56 -05:00
37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
# #
|
|
# #
|
|
|
|
#
|
|
# Notification object that produces geometry data
|
|
#
|
|
#
|
|
# SOFTWARE HISTORY
|
|
#
|
|
# Date Ticket# Engineer Description
|
|
# ------------ ---------- ----------- --------------------------
|
|
# 07/22/16 2416 tgurney Initial creation
|
|
# 09/07/17 6175 tgurney Override messageReceived
|
|
#
|
|
|
|
import dynamicserialize
|
|
from awips.dataaccess.PyNotification import PyNotification
|
|
from dynamicserialize.dstypes.com.raytheon.uf.common.dataquery.requests import RequestConstraint
|
|
|
|
class PyGeometryNotification(PyNotification):
|
|
|
|
def messageReceived(self, msg):
|
|
dataUriMsg = dynamicserialize.deserialize(msg)
|
|
dataUris = dataUriMsg.getDataURIs()
|
|
dataTimes = set()
|
|
for dataUri in dataUris:
|
|
if self.notificationFilter.accept(dataUri):
|
|
dataTimes.add(self.getDataTime(dataUri))
|
|
if dataTimes:
|
|
try:
|
|
data = self.getData(self.request, list(dataTimes))
|
|
self.callback(data)
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
|
|
def getData(self, request, dataTimes):
|
|
return self.DAL.getGeometryData(request, dataTimes)
|