aim-proxy 初次提交

This commit is contained in:
somunslotus 2024-06-21 17:13:49 +08:00
commit 12951636fb
11 changed files with 569 additions and 0 deletions

9
.idea/.gitignore vendored Normal file
View File

@ -0,0 +1,9 @@
# Default ignored files
/shelf/
/workspace.xml
# Editor-based HTTP Client requests
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml
*.xml

8
.idea/encode.iml Normal file
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" />
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

17
Dockerfile Normal file
View File

@ -0,0 +1,17 @@
FROM 172.20.32.187/pipeline-service/aimstack/aim:3.19.0
ARG http_proxy
ARG https_proxy
ENV http_proxy=${http_proxy}
ENV https_proxy=${https_proxy}
WORKDIR /app
RUN pip install httpx -i https://pypi.tuna.tsinghua.edu.cn/simple
ENV http_proxy=""
ENV https_proxy=""
COPY server.py /app/server.py
ENTRYPOINT ["python", "server.py"]

6
build-images.sh Normal file
View File

@ -0,0 +1,6 @@
#!/bin/bash
#日期精确到分钟
image=172.20.32.187/pipeline-service/aim-porxy:$(date "+%Y%m%d%H%M%S")
docker build --build-arg "http_proxy=http://172.20.32.253:3128" --build-arg "https_proxy=http://172.20.32.253:3128" -t $image -f Dockerfile .
docker push $image

40
deployment.yaml Normal file
View File

@ -0,0 +1,40 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: aim-proxy
namespace: argo
labels:
app: aim-proxy
spec:
replicas: 1
selector:
matchLabels:
app: aim-proxy
template:
metadata:
labels:
app: aim-proxy
spec:
containers:
- name: aim-proxy
image: 172.20.32.187/pipeline-service/aim-proxy:$(date "+%Y%m%d%H%M%S")
ports:
- containerPort: 7123
---
apiVersion: v1
kind: Service
metadata:
name: aim-proxy
namespace: argo
labels:
app: aim-proxy
spec:
type: NodePort
selector:
app: aim-proxy
ports:
- protocol: TCP
port: 7123
targetPort: 7123
nodePort: 30123 # 这里可以是30000到32767之间的任意值确保没有冲突

112
encode-3.19.py Normal file
View File

@ -0,0 +1,112 @@
from aim import Repo
from aim.sdk.types import QueryReportMode
from aim.sdk.sequence_collection import QueryRunSequenceCollection
# 配置 Aim repository 路径
repo_path = 'aim://172.20.32.181:30058'
# 打开 Aim repository
repo = Repo(path=repo_path)
def print_run_details(run):
if run is None:
print("Run not found.")
return
# # 打印 Run 对象的所有属性
# print("Run object attributes:")
# for attr in dir(run):
# if not attr.startswith('_'):
# print(attr)
# 打印 Run 对象的运行参数
df = run.dataframe(include_props=False)
print("all params: ", df.to_dict(orient='records'))
props = run.dataframe(include_props=True,include_params=False)
print("props: ", props.to_dict())
# 假设 Run 对象有以下属性
print(f"Run ID: {run.hash}")
print(f"Experiment: {run.experiment}")
print(f"Description: {run.description}")
print(f"Created at: {run.creation_time}")
print(f"Tags: {run.tags}")
for metric_name, metric_context, metric_run in run.iter_metrics_info():
print(f"Metric: {metric_name}, Context: {metric_context}, Run: {metric_run}")
# 通过metric_name, metric_context, metric_run获取metric的具体值
metric_value = run.get_metric(metric_name, metric_context)
print(f"{metric_name}: {metric_value}")
for metric in run.metrics():
result = metric.values.values_list()
last = metric.values.last()
print(f"{metric.name} result is :{result} ")
print(f"{metric.name}-{metric.context}: {last}")
#tt = run.collect_sequence_info(sequence_types="{'context': {'subset': 'train'}, 'name': 'loss')}")
#print("collect info is ", tt)
# # 打印所有参数
# print("\nParameters:")
# if hasattr(run, 'params'):
# for param in run.params.keys():
# print(f"{param}: {run.params[param]}")
# else:
# print("No parameters found.")
#
# # 打印所有metrics
# print("\nMetrics:")
# if hasattr(run, 'metrics'):
# for metric in run.metrics.keys():
# print(f"{metric}: {run.metrics[metric].values()}")
# else:
# print("No metrics found.")
# 打印run的详细信息
# # 查询指定实验的 runs
# experiment_name = "experiment-13"
# # runs = repo.query_runs('run.experiment=="experiment-88888"')
# runs = repo.query_runs()
# if runs is None:
# print("No runs found.")
# exit()
# else:
# print("query_runs result success.")
# result = repo.get_run("648d63882ce14525bc6ee20f")
# print_run_details(result)
# # 查询指定的 runs
# hashes_to_find = ["581c75204d134704b3c9ac27", "3dfdf9b3a4b0453790c8c7d0", "34fd862f49f64d9c8c12aa4a"]
query_res = repo.query_runs(query='run.experiment=="experiment-88888"', paginated=True).iter_runs()
#runs = repo.query_runs(query='', report_mode=QueryReportMode.PROGRESS_TUPLE)
# repo.delete_run("9cd6b4ed0e7848779274ad4f")
# 获取查询到的 runs 的 run.hash 并以数组形式返回
# query = ''
# limit = 10
# offset = 0
# x_timezone_offset = 8
# runs = QueryRunSequenceCollection(repo=repo,
# query=query,
# paginated=bool(limit),
# offset=None,
# report_mode=QueryReportMode.PROGRESS_TUPLE,
# timezone_offset=x_timezone_offset)
# count = sum(1 for _ in runs)
# print(f"Found {count} runs.")
#
runs = [item.run for item in query_res]
for run in runs:
print_run_details(run)

103
encode-3.21.py Normal file
View File

@ -0,0 +1,103 @@
from aim import Repo
from aim.sdk.types import QueryReportMode
# 配置 Aim repository 路径
repo_path = 'aim://172.20.32.181:30038'
# 打开 Aim repository
repo = Repo(repo_path)
#
# # 打印 run 的详细信息
# # def print_run_details(run):
# # if run is None:
# # print("Run not found.")
# # return
# #
# # # 打印 Run 对象的所有属性
# # print("Run object attributes:")
# # for attr in dir(run):
# # if not attr.startswith('_'):
# # print(attr)
# #
# # # 假设 Run 对象有以下属性
# # print(f"Run ID: {run.hash}")
# # print(f"Experiment: {run.experiment}")
# # print(f"Description: {run.description}")
# # print(f"Created at: {run.creation_time}")
# # print(f"Tags: {run.tags}")
# #
# # # # 打印所有参数
# # # print("\nParameters:")
# # # if hasattr(run, 'params'):
# # # for param in run.params.keys():
# # # print(f"{param}: {run.params[param]}")
# # # else:
# # # print("No parameters found.")
# # #
# # # # 打印所有metrics
# # # print("\nMetrics:")
# # # if hasattr(run, 'metrics'):
# # # for metric in run.metrics.keys():
# # # print(f"{metric}: {run.metrics[metric].values()}")
# # # else:
# # # print("No metrics found.")
# # # 打印run的详细信息
# #
#
# # 查询指定实验的 runs
# experiment_name = "experiment-13"
# #runs = repo.query_runs('run.experiment=="experiment-88888"')
# runs = repo.query_runs()
#
# if runs is None:
# print("No runs found.")
# exit()
# else:
# print("query_runs result success.")
#
# # result = repo.get_run("45311bf77ccb4ae683a7e630")
# # print_run_details(result)
# # # 查询指定的 runs
# # hashes_to_find = ["581c75204d134704b3c9ac27", "3dfdf9b3a4b0453790c8c7d0", "34fd862f49f64d9c8c12aa4a"]
# # runs = repo.query_runs(f"run.hash in {hashes_to_find}")
# # # repo.delete_run("9cd6b4ed0e7848779274ad4f")
# # # 获取查询到的 runs 的 run.hash 并以数组形式返回
# #
#
# count = list(runs).count(None)
# print(f"Found {count} runs.")
#
#
# for run in runs:
# if run is None:
# print("No runs found.")
# exit()
# else:
# print("query_runs result success.")
# print("run:", run.hash)
# # print(run.hash)
#
# print("query_runs result end.")
experiment_name = "experiment-13"
query_res = repo.query_runs(query='', paginated=True).iter_runs()
runs = [item.run for item in query_res]
for run in runs:
# filtered_runs = []
# for run in repo.iter_runs():
# if run.experiment == experiment_name:
# filtered_runs.append(run)
# print("run:", run.hash)
# # 获取所有存储的参数
# for metric_name, metric_context, metric_run in run.iter_metrics_info():
# print("metric_name:", metric_name)
# print("metric_context:", metric_context)
# print("metric_run:", metric_run)
# #获取指定参数的具体值
# metric_value = run.get_metric(metric_name, metric_context)
# print("metric_value:", metric_value)
# # 获取所有存储的图像

55
server.py Normal file
View File

@ -0,0 +1,55 @@
from fastapi import FastAPI, HTTPException
from aim.sdk.repo import Repo
import httpx
from aim.sdk.types import QueryReportMode
import traceback
import uvicorn
from datetime import datetime
#repo_path = 'aim://172.20.32.181:30058'
repo_path = 'aim://my-aim-server-service.aim:53800'
host = 'http://my-aim-ui-service.aim:43800'
# 打开 Aim repository
repo = Repo(path=repo_path)
app = FastAPI()
@app.get("/api/runs/search/run")
async def search_runs(query: str = None, paginated: bool = True, offset: str = None, limit: int = None):
try:
print("query is ", query)
print("paginated is ", paginated)
print("offset is ", offset)
print("limit is ", limit)
query_start_time = datetime.now().timestamp()
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
runs = [item.run for item in query_res]
query_end_time = datetime.now().timestamp()
query_duration = query_end_time - query_start_time
print("query time is ", query_duration)
# 根据run.creation_time 降序排序
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
results = []
http_start_time = datetime.now().timestamp()
for run in runs:
# 调用获取详细信息的接口
url = f"{host}/api/runs/{run.hash}/info"
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status()
info = response.json()
info['run_hash'] = run.hash # 将 run_hash 添加到返回的信息中
results.append(info)
if paginated:
if limit and len(results) >= limit:
break
print("result len is ", len(results))
http_end_time = datetime.now().timestamp()
http_duration = http_end_time - http_start_time
print("http time is ", http_duration)
return results
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7123)

106
server_new.py Normal file
View File

@ -0,0 +1,106 @@
from fastapi import FastAPI, HTTPException
from aim.sdk.repo import Repo
import httpx
from aim.sdk.types import QueryReportMode
from typing import Optional, Tuple, Dict, Any
from aim.sdk.repo import Repo
from aim.sdk.run import Run
from aim.web.api.runs.utils import (
checked_query,
collect_requested_metric_traces,
convert_nan_and_inf_to_str,
custom_aligned_metrics_streamer,
get_project_repo,
get_run_or_404,
get_run_params,
get_run_props,
get_run_artifacts,
metric_search_result_streamer,
run_active_result_streamer,
run_search_result_streamer,
run_logs_streamer,
run_log_records_streamer,
)
app = FastAPI()
repo_path = 'aim://172.20.32.181:30058'
# 打开 Aim repository
repo = Repo(path=repo_path)
@app.get("/api/runs/search/run")
async def search_runs(query: str, paginated: bool = True, offset: str = None, limit: int = None):
print("query is ", query)
print("paginated is ", paginated)
print("offset is ", offset)
print("limit is ", limit)
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
runs = [item.run for item in query_res]
# 根据run.creation_time 降序排序
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
results = []
print("runs len is ", len(runs))
for run in runs:
# sequence = repo.available_sequence_types()
# sequence = ["metric"]
# print("sequence is :", sequence)
# response = {
# 'params': get_run_params(run, skip_system=True),
# 'traces': run.collect_sequence_info(sequence, skip_last_value=True),
# # 'props': get_run_props(run),
# 'artifacts': get_run_artifacts(run),
# }
# # Convert NaN and Inf to strings
# response = convert_nan_and_inf_to_str(response)
#
# # response['props'].update({
# # 'notes': len(run.props.notes_obj)
# # })
# results.append(response)
re = get_run_info2(run.hash, repo, skip_system=True, sequence=("metric",))
re['run_hash'] = run.hash
results.append(re)
print("result len is ", len(results))
return results
def get_run_info2(run_id: str,
repo: Repo,
skip_system: Optional[bool] = False,
sequence: Optional[Tuple[str, ...]] = ()) -> Dict[str, Any]:
run = get_run_or_404(run_id, repo=repo)
if sequence != ():
try:
repo.validate_sequence_types(sequence)
except ValueError as e:
raise ValueError(f"Invalid sequence types: {str(e)}")
else:
sequence = repo.available_sequence_types()
response = {
'params': get_run_params(run, skip_system=skip_system),
'traces': run.collect_sequence_info(sequence, skip_last_value=False),
# 'props': get_run_props(run),
'artifacts': get_run_artifacts(run),
}
# Convert NaN and Inf to strings
response = convert_nan_and_inf_to_str(response)
# response['props'].update({
# 'notes': len(run.props.notes_obj)
# })
return response
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7124)

59
server_paralism.py Normal file
View File

@ -0,0 +1,59 @@
from fastapi import FastAPI, HTTPException
from aim.sdk.repo import Repo
import httpx
from aim.sdk.types import QueryReportMode
import asyncio
import traceback
app = FastAPI()
repo_path = 'aim://172.20.32.181:30038'
# 打开 Aim repository
repo = Repo(path=repo_path)
@app.get("/api/runs/search/run")
async def search_runs(query: str = None, paginated: bool = True, offset: str = None, limit: int = None):
try:
print("query is ", query)
print("paginated is ", paginated)
print("offset is ", offset)
print("limit is ", limit)
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
runs = [item.run for item in query_res]
# 根据run.creation_time 降序排序
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
results = []
async with httpx.AsyncClient() as client:
tasks = []
for run in runs:
url = f"http://172.20.32.181:30039/api/runs/{run.hash}/info"
tasks.append(fetch_info(client, url, run.hash))
if paginated:
if limit and len(results) >= limit:
break
infos = await asyncio.gather(*tasks)
results.extend(infos)
print("result len is ", len(results))
return results
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
async def fetch_info(client, url, run_hash):
try:
response = await client.get(url)
response.raise_for_status()
info = response.json()
info['run_hash'] = run_hash # 将 run_hash 添加到返回的信息中
return info
except httpx.HTTPStatusError as e:
# 处理 HTTP 错误
raise HTTPException(status_code=e.response.status_code, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7123)

54
train.py Normal file
View File

@ -0,0 +1,54 @@
import uuid
from aim import Run
import aim
import random
# 循环运行30次
for i in range(30):
expriment_name="experiment-0000"
# experiment_id 为随机uuid字符串
experiment_id= str(uuid.uuid4())
print("experiment_id:", experiment_id)
run = Run(repo='aim://172.20.32.181:30058', experiment=expriment_name, system_tracking_interval=None)
run['id'] = experiment_id
run["batch_size"] = 256
run["epoch"] = 10
run["learning_rate", "test"] = 0.001
run["learning_rate", "train"] = 0.002
#记录指标指标为randm数值取值为大于0小于1的小数
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=1, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=2, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=3, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=4, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=5, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=6, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=7, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=8, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=9, context={'subset': 'train'})
rand_value = round(random.uniform(0, 1), 2)
run.track(rand_value, name='loss', epoch=10, context={'subset': 'train'})
# # 创建一个 Aim 实例
# repo = aim.Repo('aim://172.20.32.181:30058') # 替换为实际的 AIM repository 路径
#
# # 迭代获取运行
# filtered_runs = []
# for run in repo.iter_runs():
# filtered_runs.append(run)
#
# # 打印或处理 filtered_runs
# for run in filtered_runs:
# print("run :" , run.hash)