Hello, I am trying out the open historian python library, the connection is successful but I’m not receiving any data points from the PMUs. Can someone please help me out? Attached is the code and it’s output.
def readTest():
# Create historian connection (the root API object)
historian = historianConnection(“10.35.0.216”)
instance: Optional[historianInstance] = Nonetry: print("Connecting to openHistorian...") historian.Connect() if not historian.IsConnected or len(historian.InstanceNames) == 0: print("No openHistorian instances detected!") else: # Get first historian instance initialInstance = historian.InstanceNames[0] print(f"Opening \"{initialInstance}\" database instance...") instance = historian.OpenInstance(initialInstance) # Get a reference to the openHistorian metadata cache historian.RefreshMetadata() metadata = historian.Metadata # Lookup measurements that represent frequency values records = metadata.GetMeasurementsBySignalType(SignalType.FREQ, instance.Name) recordCount = len(records) print(f"Queried {recordCount:,} metadata records associated with \"{instance.Name}\" database instance.") print(historian.InstanceNames) if recordCount > 0: pointIDList = metadataCache.ToPointIDList(records) # Execute a test read for data archived ten seconds ago endTime = datetime.utcnow().replace(microsecond=0, second=0) - timedelta(minutes = 5) print(endTime) startTime = endTime - timedelta(minutes = 5) print(f"Starting read for {len(pointIDList):,} points from {startTime} to {endTime}...\r\n") TestRead(instance, historian.Metadata, startTime, endTime, pointIDList) except Exception as ex: print(f"Failed to connect: {ex}") finally: if instance is not None: instance.Dispose() if historian.IsConnected: print("Disconnecting from openHistorian") historian.Disconnect()
def TestRead(instance: historianInstance, metadata: metadataCache, startTime: datetime, endTime: datetime, pointIDList: List[np.uint64]):
timeFilter = timestampSeekFilter.CreateFromRange(startTime, endTime)
pointFilter = pointIDMatchFilter.CreateFromList(pointIDList)opStart = time() reader = instance.Read(timeFilter, pointFilter) count = 0 key = historianKey() value = historianValue() while reader.Read(key, value): count += 1 print(f" Point {key.ToString(metadata)} = {value.ToString()}") print(f"\r\nRead complete for {count:,} points in {(time() - opStart):.2f} seconds.\r\n")
if name == “main”:
readTest()