From 12951636fbe1e7b8c3e79780b63295d7a4530861 Mon Sep 17 00:00:00 2001 From: somunslotus Date: Fri, 21 Jun 2024 17:13:49 +0800 Subject: [PATCH] =?UTF-8?q?aim-proxy=20=E5=88=9D=E6=AC=A1=E6=8F=90?= =?UTF-8?q?=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .idea/.gitignore | 9 ++++ .idea/encode.iml | 8 ++++ Dockerfile | 17 +++++++ build-images.sh | 6 +++ deployment.yaml | 40 ++++++++++++++++ encode-3.19.py | 112 +++++++++++++++++++++++++++++++++++++++++++++ encode-3.21.py | 103 +++++++++++++++++++++++++++++++++++++++++ server.py | 55 ++++++++++++++++++++++ server_new.py | 106 ++++++++++++++++++++++++++++++++++++++++++ server_paralism.py | 59 ++++++++++++++++++++++++ train.py | 54 ++++++++++++++++++++++ 11 files changed, 569 insertions(+) create mode 100644 .idea/.gitignore create mode 100644 .idea/encode.iml create mode 100644 Dockerfile create mode 100644 build-images.sh create mode 100644 deployment.yaml create mode 100644 encode-3.19.py create mode 100644 encode-3.21.py create mode 100644 server.py create mode 100644 server_new.py create mode 100644 server_paralism.py create mode 100644 train.py diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..7ff76fd --- /dev/null +++ b/.idea/.gitignore @@ -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 \ No newline at end of file diff --git a/.idea/encode.iml b/.idea/encode.iml new file mode 100644 index 0000000..d0876a7 --- /dev/null +++ b/.idea/encode.iml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..b753926 --- /dev/null +++ b/Dockerfile @@ -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"] \ No newline at end of file diff --git a/build-images.sh b/build-images.sh new file mode 100644 index 0000000..a395042 --- /dev/null +++ b/build-images.sh @@ -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 \ No newline at end of file diff --git a/deployment.yaml b/deployment.yaml new file mode 100644 index 0000000..0a2a5ad --- /dev/null +++ b/deployment.yaml @@ -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之间的任意值，确保没有冲突 diff --git a/encode-3.19.py b/encode-3.19.py new file mode 100644 index 0000000..25da33e --- /dev/null +++ b/encode-3.19.py @@ -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) \ No newline at end of file diff --git a/encode-3.21.py b/encode-3.21.py new file mode 100644 index 0000000..6dea792 --- /dev/null +++ b/encode-3.21.py @@ -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) +# # 获取所有存储的图像 + diff --git a/server.py b/server.py new file mode 100644 index 0000000..69d84c5 --- /dev/null +++ b/server.py @@ -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) diff --git a/server_new.py b/server_new.py new file mode 100644 index 0000000..8a9bc4d --- /dev/null +++ b/server_new.py @@ -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) + diff --git a/server_paralism.py b/server_paralism.py new file mode 100644 index 0000000..8a4d679 --- /dev/null +++ b/server_paralism.py @@ -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) diff --git a/train.py b/train.py new file mode 100644 index 0000000..6d7dc34 --- /dev/null +++ b/train.py @@ -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) \ No newline at end of file