Python编程语言的基本实用程序模块
pbu的Python项目详细描述
Python基本实用程序pbu
在PyPi提供
目录
- Installation
- Usage
- Classes
- JSON-一个类似JavaScript的字典访问帮助程序
- Logger-Python日志框架的包装器
- TimeSeries-功能强大的帮助类来组织时间序列
- AbstractMongoStore-MongoDB访问的帮助器和包装器类
- AbstractMysqlStore-MySQL访问的帮助器和包装器类
- BasicMonitor-监视类编排常规操作
- ConstantListing-允许从常量类获取属性值的父类
- Functions 在
安装
通过pip安装：
pip install pbu
使用
可选：如果您有要求.txt文件，您可以添加pbu：
然后，只需导入所需的类/模块：
frompbuimportJSON# and start using itobj=JSON({"my":{"obj":"content"}})print(obj.my.obj)
班级
JSON
这是本机dict类的改编，使用“点表示法”提供类似Javascript的字典访问
（例如person.relations[0].address.street）而不是Python本机括号表示法（例如。
person["relations"][0]["address"]["street"]）。它重写基本的__getattr__和__setattr__方法作为
管理词典内容的快捷方式。在
示例
frompbuimportJSONmy_obj=JSON({"initial":"content"})print(my_obj.initial)# prints out "content"my_obj.initial={"a":5,"b":3}print(my_obj.initial.a+my_obj.initial.b)# prints out 8my_obj.initial.b=13print(my_obj.initial.a+my_obj.initial.b)# prints out 18my_obj.extension=10print(my_obj.extension)# prints out 10
记录器
这是一个允许写入日志文件的基本记录器，因为logger.info它将调试日志对于logger.error或
logger.exception它写一个错误.log文件。在
示例
frompbuimportLoggerlogger=Logger(name="logger-name")logger.debug("Some debug message goes here")logger.error("Error executing something")logger=Logger(name="logger-name",log_folder="./logs")logger.debug("This will create the debug.log and error.log in the ./logs folder")
时间序列
时间序列类是一个帮助工具，它允许编译复杂的时间序列，提供添加功能 时间序列，删除时间序列，最重要的是将时间序列与先前定义的时间戳对齐 通过插值缺失值并在规定时间内重新校准测量值来解决问题 系列。在
它支持两种不同的结构：
字典项列表
fromdatetimeimportdatetime,timedeltalist_of_dict=[{"date_time":datetime.now(),"measurement_1":12,"measurement_2":15},{"date_time":datetime.now()+timedelta(hours=1),"measurement_1":10,"measurement_2":16},{"date_time":datetime.now()+timedelta(hours=2),"measurement_1":9,"measurement_2":12},]
列表字典
fromdatetimeimportdatetime,timedeltadict_of_list={"date_time":[datetime.now(),datetime.now()+timedelta(hours=1),datetime+timedelta(hours=2)],"measurement_1":[12,10,16],"measurement_2":[15,16,12],}
示例
frompbuimportTimeSeriesfromdatetimeimportdatetime,timedelta# initial time series base data (you can add measurements as well or provide as list of dictionariesdict_of_list={"date_time":TimeSeries.create_date_range(datetime.now(),datetime.now()+timedelta(days=1),timedelta(hours=3)),}# init time seriests=TimeSeries(input_data=dict_of_list,date_time_key="date_time")# add values (ensure same length as date_time series)ts.add_values("measurement_1",[12,10,16,10,5,8,12,9])# you can translate into a list of dictionary items (keys are maintained)list_of_dict=ts.translate_to_list_of_dicts()# extract data series from the time seriesmeasurement_1=ts.get_values("measurement_1")# create new series that provides same value for all timestampsts.fill_values("constant_series",5)# remove a series from the total data structurets.remove_series("constant_series")# re-sample data to 5 minute resolution, interpolating values, also pre-pending another day in front of the time series ts.align_to_resolution(resolution=timedelta(minutes=5),start_date=datetime.now()-timedelta(days=1))# this will result in "interpolated" values for the first day, using the first value (12) to fill missing valuesprint(len(ts.translate_to_list_of_dicts()))# 12 an hour, 2 days, 48 * 12 = ~576 items# the same can also be achieved by:ts.set_resolution(timedelta(minutes=5))# no need to provide resolution nowts.align_to_resolution(start_date=datetime.now()-timedelta(days=1))
抽象Mongostore
数据库存储，其中包含用于访问MongoDB的助手函数。每个存储实例代表一个集合。这个
附带了一个AbstractMongoDocument类，可用于对存储在MongoDB中的文档类型进行建模
收藏。在
示例
frompbuimportAbstractMongoStore,AbstractMongoDocument# this is the object type stored in the mongo storeclassMyObjectType(AbstractMongoDocument):def__init__(self,val1,val2):# optional: provide id and data model version super().__init__()self.attribute=val1self.attribute2=val2,defto_json(self):# init with version and idresult=super().to_json()# add attributes to dictionary and returnresult["attribute"]=self.attributeresult["attribute2"]=self.attribute2returnresult@staticmethoddeffrom_json(json):result=MyObjectType(json["attribute1"],json["attribute2"])# get _id and version attributesresult.extract_system_fields(json)returnresultclassMyObjectStore(AbstractMongoStore):def__init__(self,mongo_url,db_name,collection_name,data_model_version):# provide object type class as de-serialisation class (providing from_json and to_json)super.__init__(mongo_url,db_name,collection_name,MyObjectType,data_model_version)# create instance of storestore=MyObjectStore("mongodb://localhost:27017","mydb","colName",5)# create document using a dictionarystore.create({"version":5,"attribute1":"a","attribute2":16,})# or use the typedoc=MyObjectType("a",16)doc.version=5doc_id=store.create(doc)# update single document using helper functionsstore.update(AbstractMongoStore.id_query(doc_id),AbstractMongoStore.set_update(["attribute1","attribute2"],["b",12]))# returns a list of MyObjectType objects matching the versionlist_of_results=store.query({"version":5})
基本监视器
一个抽象类，为运行监视器提供基本功能，监视器是在常规程序中运行特定例程的线程 间隔时间。它可以是等待处理新任务的执行器（每5秒检查一次），也可以是 定期监控一些读数。监视器被包装以在出现错误时重新启动。在
示例
frompbuimportBasicMonitorclassMyOwnMonitor(BasicMonitor):def__init__(self,data):super().__init__(monitor_id="my_id",wait_time=5)# waits 5 seconds between each execution loopself.data=datadefrunning(self):whileself.active:# your code goes here (example):# result = fetch_data(self.data)# store_result(result)self.wait()
如果要定期运行，running方法需要稍作修改：
fromtimeimporttimefrompbuimportBasicMonitorclassMyRegularOwnMonitor(BasicMonitor):def__init__(self,data):super().__init__(monitor_id="another_id",wait_time=60,run_interval=True)# execute every 60 secondsself.data=datadefrunning(self):whileself.active:start_ts=time()# capture start of loop# your code goes here (example):# result = do_something(self.data)# store_result(result)self.wait(exec_duration=round(time()-start_ts))# include the execution duration
也可以将自定义记录器作为custom_logger参数传递给构造函数。默认情况下，它将使用
pbu.Logger并记录主要事件，如启动/停止/重新启动和错误。在
管理并运行监视器
importthreadingdefstart_monitor_thread(monitor):""" Thread function to be run by the new thread. :param monitor: BasicMonitor - an instance of sub-class of BasicMonitor """# start the monitormonitor.start()# create monitor instance of your own class that implements BasicMonitorregular_monitor=MyRegularOwnMonitor(data={"some":"data"})# create thread with start-up function and start itt=threading.Thread(target=start_monitor_thread,args=(regular_monitor,),daemon=True)t.start()# in a separate piece of code (e.g. REST handler or timer) you can stop the monitor instanceregular_monitor.stop()
停止监视器不会中断当前线程。例如，如果监视器处于等待期，并且您发送
stop信号，线程仍将运行，直到等待期结束。在
In an API scenario, I recommend using a
dictorlistto cache monitors and retrieve them via the API using theto_json()method for identification. This then allows you to signal starting / stopping of monitors by providing the monitor ID and lookup the monitor instance in the monitor cache.
BasicMonitor方法
start()-启动监视器stop()-停止监视器to_json()-返回一个包含基本监视器技术信息（id、状态、等待行为等）的字典wait_till_midnight()-在机器时区中等待下一个午夜wait(exec_duration=0)-等待构造函数中指定的时间，如果是run_interval=True，则等待 可选exec_duration，如果提供的话。在
恒定列表
管理常量是避免打字错误的好方法。想象一下下面的课程：
classTags:GEO="GEO"EQUIPMENT="EQUIPMENT"
这允许您做：Tags.GEO允许您使用IDEs自动完成，避免打字错误。
但是如果您想以编程方式获取all可能的Tags的值，可以使用pbu的ConstantListing类：
frompbuimportConstantListingclassTags(ConstantListing):GEO="GEO"EQUIPMENT="EQUIPMENT"list_of_values=Tags().get_all()# will return ['GEO', 'EQUIPMENT']
功能
list_to_json
frompbuimportlist_to_json# assuming we have `my_store` as an instance of MongoDB store or MySQL store, you can:list_of_dictionaries=list_to_json(item_list=my_store.get_all())# output is a list of dictionaries
此函数对继承自AbstractMongoDocument或AbstractMysqlDocument和
使用传入函数的任何对象的to_json()方法将它们转换为字典。传入的对象
函数需要to_json()方法，并需要返回对象的字典表示。这个
函数只是一个映射快捷方式。在
default_options
frompbuimportdefault_optionsDEFAULTS={"a":1,"b":2,"c":3,}result=default_options(default=DEFAULTS,override={"b":4,"d":5})# result is: {"a": 1, "b": 4, "c": 3, "d": 5}
如果要避免默认键之外的其他键，可以提供第三个参数：
frompbuimportdefault_optionsDEFAULTS={"a":1,"b":2,}result=default_options(default=DEFAULTS,override={"b":4,"d":5},allow_unknown_keys=False)# result is: {"a": 1, "b": 4}
default_value
frompbuimportdefault_valueresult=default_value(value=None,fallback=5)# None is by default disallowed# result is 5result=default_value(value=0,fallback=5,disallowed=[None,0])# either 0 or None would return the fallback# result is 5result=default_value(0,5)# value will be used, as it doesn't match None# result is 0
- 项目
标签：
欢迎加入QQ群-->： 979659372
