Python一些可能用的到的函数系列130 UCS-Time Brick

说明

UCS对象是基于GFGoLite进行封装,且侧重于实现UCS规范。

内容

1 函数

我发现pydantic真是一个特别好用的东西,可以确保在数据传递时的可靠,以及对某个数据模型的描述。

以下,UCS给出了id、time相关的brick映射,并给出了时间到字符的转换。其方便之处在于,本地几乎不需要管怎么实现的,只要有服务支持即可。效率也足够高。

from typing import List, Optional
from pydantic import BaseModelimport requests as req 
class UCS(BaseModel):__version__ =1.1gfgo_lite_server: str = 'http://172.17.0.1:24090/'def get_brick_name(self, some_id = None):some_dict = {}some_dict['rec_id'] = some_idurl = self.gfgo_lite_server + 'get_brick_name/'res = req.post(url, json = some_dict).json()return res def get_brick_name_s(self, some_id_list = None):some_dict = {}some_dict['rec_id_list'] = some_id_listurl = self.gfgo_lite_server + 'get_brick_name_s/'res = req.post(url, json = some_dict).json()return res def gfgo(self, kwargs = None, pack_func = None):url = self.gfgo_lite_server + 'gfgo/'return req.post(url, json = {'kwargs':kwargs,'pack_func':pack_func}).json()def dt_str2num(self, some_dt_str = None):some_dict = {}some_dict['some_dt_str'] = some_dt_strurl = self.gfgo_lite_server + 'str2num/'res = req.post(url, json = some_dict).json()return  resdef dt_str2num_s(self, some_dt_str_list = None):some_dict = {}some_dict['some_dt_str_list'] = some_dt_str_listurl = self.gfgo_lite_server + 'str2num_s/'res = req.post(url, json = some_dict).json()return  res# 时间名称def get_time_brick_name(self, dt_str_or_ts = None):some_dict = {}some_dict['dt_str_or_ts'] = dt_str_or_tsurl = self.gfgo_lite_server + 'get_time_brick_name/'res = req.post(url, json = some_dict).json()return res def get_time_brick_name_s(self, dt_str_or_ts_list = None):some_dict = {}some_dict['dt_str_or_ts_list'] = dt_str_or_ts_listurl = self.gfgo_lite_server + 'get_time_brick_name_s/'res = req.post(url, json = some_dict).json()return res # 1.1 >>>def get_brick_list(self, start_brick_name = None, end_brick_name = None):some_dict = {}some_dict['start_brick_name'] = start_brick_namesome_dict['end_brick_name'] = end_brick_nameurl = self.gfgo_lite_server + 'get_brick_list/'res = req.post(url, json = some_dict).json()return res def get_brick_bounds(self, brick_name = None):some_dict = {}some_dict['brick_name'] = brick_nameurl = self.gfgo_lite_server + 'get_brick_bounds/'res = req.post(url, json = some_dict).json()return res # 1.2 >>> 时间# get_time_brick_bounds_s - 待def get_time_brick_bounds_s(self, brick_name_list = None, char_or_num = 'char'):some_dict = {}some_dict['brick_name_list'] = brick_name_listsome_dict['char_or_num'] = char_or_numurl = self.gfgo_lite_server + 'get_time_brick_bounds_s/'res = req.post(url, json = some_dict).json()return res # get_time_brick_listdef get_time_brick_list(self,start_brick_name = None, end_brick_name = None):some_dict = {}some_dict['start_brick_name'] = start_brick_namesome_dict['end_brick_name'] = end_brick_nameurl = self.gfgo_lite_server + 'get_time_brick_list/'res = req.post(url, json = some_dict).json()return res 

2 应用

2.1 时间转换

ucs = UCS(gfgo_lite_server = 'http://xxx:24090/')
ucs.dt_str2num('2024-01-31 11:11:11')
1706670671ucs.dt_str2num_s(['2024-01-31 11:11:11','2024-02-01 11:11:11' ])
[1706670671, 1706757071]

反函数可以用get_time_str1

In [19]: get_time_str1(1706670671)
Out[19]: '2024-01-31 11:11:11'

2.2 UCS-time

一个简单的应用如下:我需要制定一个静态数据同步计划,计划需要随着时间轴,按照时间块熟顺序向前推进。具体来说,这个对于数据库迁移(合并)与回测都有关系。

tbrick1 = ucs.get_time_brick_name('2024-01-31 11:11:11')
'2024.01.31.11'
tbrick2 = ucs.get_time_brick_name('2024-02-10 11:11:11')
'2024.02.10.11'tbrick_list = ucs.get_time_brick_list(start_brick_name = tbrick1, end_brick_name = tbrick2)
['2024.01.31.11', '2024.01.31.12', '2024.01.31.13', ...]ucs.get_time_brick_bounds_s(tbrick_list[:3])
[['2024-01-31 11:00:00', '2024-01-31 12:00:00'],['2024-01-31 12:00:00', '2024-01-31 13:00:00'],['2024-01-31 13:00:00', '2024-01-31 14:00:00']]

btw, 在探索完UCS-time之后,我发现大约因为时间观念比较深入人心,其实是可以更自由的转换的,

# 将一般字符串转为UCS 名称
def dt_str2ucs_blockname(some_dt_str):some_dt_str1   =some_dt_str.replace('-','.').replace(' ','.').replace(':','.')return '.'.join(some_dt_str1.split('.')[:4])'''
dt_str2ucs_blockname('2024-06-24 09:30:00')
'2024.06.24.09'
'''

这些函数也被纳入基础函数,有时候可以起到辅助的作用,特别在使用GFGolite不那么方便的时候。

一个更具体的场景,我觉得是具有通用性的,描述如下:

  • 1 数据没有顺序id,但是具备入库时间(create_time)
  • 2 数据入库时间是不连续的,有一些时间段密集,更多的时间可能空闲,但也并不是一条没有。
  • 3 每次执行时,worker总是能够有效推进一个brick

具体做法:

p01: 将需要的时间块按序持久化为本地文件,并将对应的bounds作为字典

p02: 获取当前时间的时间块

# 将标准时间字符串转为brick
def timestr2timeblock(some_time_str = None):x1 = some_time_str.replace('-','.').replace(' ','.').replace(':','.')return '.'.join(x1.split('.')[:4])the_time_now = get_time_str1()
the_time_brick = timestr2timeblock(the_time_now)

p03:从gb中获取上一次处理过的最新brick,并锁定在时间轴的位置。


worker_buffer_space = 'sp_worker.general'
tier1 = 'some_tier'
tier2 = 'ucs_time_brick_ordered.sniffer'
prefix = '.'.join([worker_buffer_space,tier1,tier2]) +'.'# ==========================  Load
time_brick_list = from_pickle('time_brick_list')
time_bounds_dict = from_pickle('time_bounds_dict')
gb = GlobalBuffer()
watch_brick_pos = time_brick_list.index(the_time_brick)

p04:按照已处理之后,当前观察时间之前的原则,选出available_time_brick_list

latest_data_brick = gb.getx(prefix +'last_brick_handled')
if latest_data_brick is None:start_pos = 0
else:start_pos =  time_brick_list.index(latest_data_brick)available_time_brick_list = time_brick_list[start_pos+1:watch_brick_pos]

p05:按顺序遍历时间块,直到处理到一个有效的brick才中断本次循环

# cur_data_brick = available_time_brick_list[0]
for cur_data_brick in available_time_brick_list:print(cur_data_brick)cur_bound = time_bounds_dict[cur_data_brick]print(cur_bound)resp = 【获取待处理数据】if len(resp):res_df = pd.DataFrame(resp, columns = keep_cols)if ok_to_handle:【将结果分批删除】listofdict2 = slice_list_by_batch2(xxx.to_dict(orient='records'), 1000)for some_listofdict in listofdict2:【更新已处理块信息】gb.setx(prefix +'last_brick_handled',cur_data_brick, persist=True)break

通过循环或者定时任务唤起这样的worker就可以。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/41562.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【分布式系统五】监控平台Zabbix实际监控运用(命令+截图详细版)

目录 一.Zabbix 监控 Windows 1.安装zabbix 2.Web 页面添加主机,关联模板 二.Zabbix 监控 Java 应用 1.安装tomcat 2.服务端安装 zabbix-java-gateway 3.Web 页面添加主机,关联模板 三.Zabbix 监控 SNMP 1.服务端安装 snmp 监控程序 2.修改 sn…

RT-Thread和freeRTOS启动流程

一. freeRTOS启动流程 二. RT-Thread启动流程 因为RT-Thread中我们定义了补丁函数也叫做钩子函数--$Sub$$main()--作为一个新功能函数,可以将原有函数劫持下来,并在之后的程序运行中加上$Super $ $前缀来重新调用原始函数。 所以启动流程是$Sub$$main(…

Chiasmodon:一款针对域名安全的公开资源情报OSINT工具

关于Chiasmodon Chiasmodon是一款针对域名安全的公开资源情报OSINT工具,该工具可以帮助广大研究人员从各种来源收集目标域名的相关信息,并根据域名、Google Play应用程序、电子邮件地址、IP地址、组织和URL等信息进行有针对性的数据收集。 该工具可以提…

利用node连接mongodb实现一个小型后端服务系统demo

http 请求 实现get请求数据库数据;实现添加数据实现编辑数据实现删除数据实现导出txt文件、Excel文件实现查询数据库数据并利用导出为excel文件 node 版本 16.16.0 node 版本 18.16.0 会连接 MongoDB 数据库错误。 Connected to MongoDB failed MongoServerSele…

Nginx-简介

介绍 nginx是一款HTTP和反向代理服务器、邮件代理服务器和通用TCP/IP代理服务器,在俄罗斯广泛使用,用于代理高负载站点。 版本 nginx开源版nginx plus企业版openresty将nginx和lua脚本结合 tengine更稳定、高性能 正向代理 客户端和代理服务是一伙的…

【vue动态组件】VUE使用component :is 实现在多个组件间来回切换

VUE使用component :is 实现在多个组件间来回切换 component :is 动态父子组件传值 相关代码实现&#xff1a; <component:is"vuecomponent"></component>import componentA from xxx; import componentB from xxx; import componentC from xxx;switch(…

生产力工具|viso常用常见科学素材包

一、科学插图素材网站 一图胜千言&#xff0c;想要使自己的论文或重要汇报更加引人入胜&#xff1f;不妨考虑利用各类示意图和科学插图来辅助研究工作。特别是对于新手或者繁忙的科研人员而言&#xff0c;利用免费的在线科学插图素材库&#xff0c;能够极大地节省时间和精力。 …

Python字符编码检测利器: chardet库详解

Python字符编码检测利器: chardet库详解 1. chardet简介2. 安装3. 基本使用3.1 检测字符串编码3.2 检测文件编码 4. 高级功能4.1 使用UniversalDetector4.2 自定义编码检测 5. 实际应用示例5.1 批量处理文件编码5.2 自动转换文件编码 6. 性能优化7. 注意事项和局限性8. 总结 在…

【代码随想录】【算法训练营】【第58天】 [卡码110]字符串接龙 [卡码105]有向图的完全可达性 [卡码106]岛屿的周长

前言 思路及算法思维&#xff0c;指路 代码随想录。 题目来自 卡码网。 day 59&#xff0c;周五&#xff0c;继续ding~ 题目详情 [卡码110] 字符串接龙 题目描述 卡码110 字符串接龙 解题思路 前提&#xff1a; 思路&#xff1a; 重点&#xff1a; 代码实现 C语言 […

Jackson库使用教程

1. Jackson概述 定义: Jackson是一个基于Java的开源JSON解析工具&#xff0c;用于Java对象与JSON数据的互相转换。示例JSON:{"author": "一路向北_Coding","age": 20,"hobbies": ["coding", "leetcode", "r…

昇思25天学习打卡营第13天|linchenfengxue

Diffusion扩散模型 关于扩散模型&#xff08;Diffusion Models&#xff09;有很多种理解&#xff0c;本文的介绍是基于denoising diffusion probabilistic model &#xff08;DDPM&#xff09;&#xff0c;DDPM已经在&#xff08;无&#xff09;条件图像/音频/视频生成领域取得…

小蜜蜂WMS与小蜜蜂WMS对接集成根据条件获取客户信息列表(分页)连通新增客户信息(小蜜蜂读写测试)

小蜜蜂WMS与小蜜蜂WMS对接集成根据条件获取客户信息列表&#xff08;分页&#xff09;连通新增客户信息(小蜜蜂读写测试) 接通系统&#xff1a;小蜜蜂WMS 天津市小蜜蜂计算机技术有限公司&#xff08;acbee&#xff0c;TianJinACBEEComputerTechnologyCo.,Ltd&#xff09;成立于…

基于图像处理的滑块验证码匹配技术

滑块验证码是一种常见的验证码形式&#xff0c;通过拖动滑块与背景图像中的缺口进行匹配&#xff0c;验证用户是否为真人。本文将详细介绍基于图像处理的滑块验证码匹配技术&#xff0c;并提供优化代码以提高滑块位置偏移量的准确度&#xff0c;尤其是在背景图滑块阴影较浅的情…

上海市计算机学会竞赛平台2023年2月月赛丙组平分数字(一)

题目描述 给定 &#x1d45b;n 个整数&#xff1a;&#x1d44e;1,&#x1d44e;2,⋯ ,&#x1d44e;&#x1d45b;a1​,a2​,⋯,an​&#xff0c;请判定能否将它们分成两个部分&#xff08;不得丢弃任何数字&#xff09;&#xff0c;每部分的数字之和一样大。 输入格式 第…

模拟,CF 570C - Replacement

一、题目 1、题目描述 2、输入输出 2.1输入 2.2输出 3、原题链接 570C - Replacement 二、解题报告 1、思路分析 1、长为cnt的连续串的最小操作次数为cnt - 1 2、每次将一个非. 替换为. f要么增加1要么增加2 只有前后都是 . 的时候会增加2 同理&#xff0c;当我们将一…

STM32外扩SRAM及用法

一.概述 一般单片机有片内的RAM&#xff0c;但都不多&#xff0c;比如&#xff1a;STM32F407ZGT6 自带了 192K 字节的 RAM&#xff0c;对一般应用来说&#xff0c;已经足够了&#xff0c;不过在一些对内存要求高的场合&#xff0c;比如做华丽效果的 GUI&#xff0c;处理大量数据…

swagger的接口文档导入到yapi上

一、访问swagger接口 swagger集成到项目后&#xff0c;通过http:\\ip:port/swagger-ui.html 访问。 说明&#xff1a;这里的路径是基于swagger2。如果用swagger3&#xff0c;需要用swagger3的路径进行访问。 访问如图&#xff1a; 这就是swagger接口首页。如果想导入到yapi上…

module_param_named 内核启动时模块参数实现原理

基于上节内核启动参数实现原理内容, 其中对early_param的实现流程做了分析, 已基本清晰. 但有不少的参数是在内核模块中声明的, 具体赋值流程也值得一探究竟. nomodeset 装过Linux系统的同学可能多少有看到过nomodeset这个参数, 解决一些显卡点不亮Linux的问题. 那么这个nomo…

AI绘画Stable Diffusion 新手入门教程:万字长文解析Lora模型的使用,快速上手Lora模型!

大家好&#xff0c;我是设计师阿威 今天给大家讲解一下AI绘画Stable Diffusion 中的一个重要模型—Lora模型&#xff0c;如果还有小伙伴没有SD安装包的&#xff0c;可以看我往期入门教程2024最新超强AI绘画Stable Diffusion整合包安装教程&#xff0c;零基础入门必备&#xff…

React Hooks --- 分享自己开发中常用的自定义的Hooks (1)

为什么要使用自定义 Hooks 自定义 Hooks 是 React 中一种复用逻辑的机制&#xff0c;通过它们可以抽离组件中的逻辑&#xff0c;使代码更加简洁、易读、易维护。它们可以在多个组件中复用相同的逻辑&#xff0c;减少重复代码。 1、useThrottle 代码 import React,{ useRef,…