PyPI 官网下载 | ICS_IPA-0.0.1.4.tar.gz

# This file was automatically generated by SWIG (http://www.swig.org). # Version 3.0.12 # # Do not make changes to this file unless you know what you are doing--modify # the SWIG interface file instead. from sys import version_info as _swig_python_version_info if _swig_python_version_info >= (2, 7, 0): def swig_import_helper(): import importlib pkg = __name__.rpartition('.')[0] mname = '.'.join((pkg, '_DataFileIOLibrary')).lstrip('.') try: return importlib.import_module(mname) except ImportError: return importlib.import_module('_DataFileIOLibrary') _DataFileIOLibrary = swig_import_helper() del swig_import_helper elif _swig_python_version_info >= (2, 6, 0): def swig_import_helper(): from os.path import dirname import imp fp = None try: fp, pathname, description = imp.find_module('_DataFileIOLibrary', [dirname(__file__)]) except ImportError: import _DataFileIOLibrary return _DataFileIOLibrary try: _mod = imp.load_module('_DataFileIOLibrary', fp, pathname, description) finally: if fp is not None: fp.close() return _mod _DataFileIOLibrary = swig_import_helper() del swig_import_helper else: import _DataFileIOLibrary del _swig_python_version_info try: _swig_property = property except NameError: pass # Python < 2.2 doesn't have 'property'. try: import builtins as __builtin__ except ImportError: import __builtin__ def _swig_setattr_nondynamic(self, class_type, name, value, static=1): if (name == "thisown"): return self.this.own(value) if (name == "this"): if type(value).__name__ == 'SwigPyObject': self.__dict__[name] = value return method = class_type.__swig_setmethods__.get(name, None) if method: return method(self, value) if (not static): if _newclass: object.__setattr__(self, name, value) else: self.__dict__[name] = value else: raise AttributeError("You cannot add attributes to %s" % self) def _swig_setattr(self, class_type, name, value): return _swig_setattr_nondynamic(self, class_type, name, value, 0) def _swig_getattr(self, class_type, name): if (name == "thisown"): return self.this.own() method = class_type.__swig_getmethods__.get(name, None) if method: return method(self) raise AttributeError("'%s' object has no attribute '%s'" % (class_type.__name__, name)) def _swig_repr(self): try: strthis = "proxy of " + self.this.__repr__() except __builtin__.Exception: strthis = "" return "<%s.%s; %s >" % (self.__class__.__module__, self.__class__.__name__, strthis,) try: _object = object _newclass = 1 except __builtin__.Exception: class _object: pass _newclass = 0 def GetNumChannels(pJsonFile): return _DataFileIOLibrary.GetNumChannels(pJsonFile) GetNumChannels = _DataFileIOLibrary.GetNumChannels def OpenDataFile(pDataFile, pJsonFile): return _DataFileIOLibrary.OpenDataFile(pDataFile, pJsonFile) OpenDataFile = _DataFileIOLibrary.OpenDataFile def GetMeasurementTimeBounds(indatapointer): return _DataFileIOLibrary.GetMeasurementTimeBounds(indatapointer) GetMeasurementTimeBounds = _DataFileIOLibrary.GetMeasurementTimeBounds def SetActiveMask(indatapointer, pMask): return _DataFileIOLibrary.SetActiveMask(indatapointer, pMask) SetActiveMask = _DataFileIOLibrary.SetActiveMask def JumpBeforeTimestamp(indatapointer, dTime): return _DataFileIOLibrary.JumpBeforeTimestamp(indatapointer, dTime) JumpBeforeTimestamp = _DataFileIOLibrary.JumpBeforeTimestamp def JumpAfterTimestamp(indatapointer, dTime): return _DataFileIOLibrary.JumpAfterTimestamp(indatapointer, dTime) JumpAfterTimestamp = _DataFileIOLibrary.JumpAfterTimestamp def GetNextRecord(indatapointer): return _DataFileIOLibrary.GetNextRecord(indatapointer) GetNextRecord = _DataFileIOLibrary.GetNextRecord def CloseDataFile(indatapointer): return _DataFileIOLibrary.CloseDataFile(indatapointer) CloseDataFile = _DataFileIOLibrary.CloseDataFile def CreateDatabase(pMdfFile, pDbFile): return _DataFileIOLibrary.CreateDatabase(pMdfFile, pDbFile) CreateDatabase = _DataFileIOLibrary.CreateDatabase def CreateDatabaseForSignals(pMdfFile, pJsonFile, pDbFile): return _DataFileIOLibrary.CreateDatabaseForSignals(pMdfFile, pJsonFile, pDbFile) CreateDatabaseForSignals = _DataFileIOLibrary.CreateDatabaseForSignals def GetChannels(pMdfFile, pJsonFile): return _DataFileIOLibrary.GetChannels(pMdfFile, pJsonFile) GetChannels = _DataFileIOLibrary.GetChannels def GetChannelStatistics(pMdfFile, pJsonFile): return _DataFileIOLibrary.GetChannelStatistics(pMdfFile, pJsonFile) GetChannelStatistics = _DataFileIOLibrary.GetChannelStatistics import sys import os import numpy as np import pprint import json class CreateDSR: def __init__(self): self.dsr = {"HitLists" : []} def Begin(self, data, hitDiscretion = "Hit number ", initTrigger=False): self.numRec = 0 self.triggered = initTrigger self.hitDiscretion = hitDiscretion curTimestamp = data.JumpAfterTimestamp(0) self.data = data self.dsr["HitLists"].append({"FilenameAndPath": data.dbFileName}) def IncludeRecord(self, included): if included: if not self.triggered: if "Hits" not in self.dsr["HitLists"][-1]: self.dsr["HitLists"][-1]["Hits"] = [] self.dsr["HitLists"][-1]["Hits"].append({"Description" : self.hitDiscretion + str(self.numRec), "StartTimes": self.data.RecordTimestamp }) self.numRec += 1 self.triggered = True elif self.triggered: self.dsr["HitLists"][-1]["Hits"][-1]["EndTime"] = self.data.RecordTimestamp self.triggered = False def End(self): if self.triggered: self.dsr["HitLists"][-1]["Hits"][-1]["EndTime"] = data.GetMeasurementTimeBounds()[2] - data.GetMeasurementTimeBounds()[1] def Add(self, data, callback, hitDiscretion = "Hit number ", initTrigger=False): ''' the AddToDSR function takes two arguments the first being ICSData class and the second being a function with two paramaters as an argument The AddToDSR calls the function for every data point it iterates though to determan if it should be included in the DSR file ''' numRec = 0 triggered = initTrigger dsr = self.dsr curTimestamp = data.JumpAfterTimestamp(0) dsr["HitLists"].append({"FilenameAndPath": data.dbFileName}) while curTimestamp != sys.float_info.max: if callback(data.GetPoints(), data.GetTimeStamps()): if not triggered: if "Hits" not in dsr["HitLists"][-1]: dsr["HitLists"][-1]["Hits"] = [] dsr["HitLists"][-1]["Hits"].append({"Description" : hitDiscretion + str(numRec), "StartTimes": curTimestamp }) numRec += 1 triggered = True elif triggered: dsr["HitLists"][-1]["Hits"][-1]["EndTime"] = curTimestamp triggered = False curTimestamp = data.GetNextRecord() if triggered: dsr["HitLists"][-1]["Hits"][-1]["EndTime"] = data.GetMeasurementTimeBounds()[2] - data.GetMeasurementTimeBounds()[1] def save(self, filename = "data.dsr"): with open(filename, 'w') as outfile: json.dump(self.dsr, outfile) class ICSDataFile: def __init__(self, dbFileName, jsonFileName, AutoCleanUpTempFiles = True): self.UsingTempFile = False if len(dbFileName) > 0: name, extension = os.path.splitext(dbFileName) if extension.lower() != ".db": if not os.path.isfile(name + ".db"): CreateDatabaseForSignals(dbFileName, jsonFileName, name + ".db") self.UsingTempFile = True dbFileName = name + ".db" self.dbFileName = dbFileName self.jsonFileName = jsonFileName