ignore .coding

backend/.coding/functions.py

@@ -1,61 +1,116 @@
-import requests
+import os
+import pandas
+import glob
 
-def retrieval_from_knowledge_base(
-        query: str,
-        topk: int
-        ) -> str:
-    """
-    Retrieval for knowledge from the knowledge base based on the specified query and returns the topk results.
-    
-    Parameters:
-    query (str): The query for knowledge retrieval.
-    topk (int): The number of top results to return, default is 3.
-    
-    Returns:
-    str: The result of the knowledge retrieval in JSON format.
-    """
-    url = 'http://127.0.0.1:7080/v1/chat-messages'
-    headers = {
-        'Authorization': f'Bearer app-uJgo3TQKcS1O9PMCDHko71Fp',
-        'Content-Type': 'application/json'
-    }
-    data = {
-        "inputs": {"topK": topk},
-        "query": query,
-        "response_mode": "blocking",
-        "user": "tangger",
-        "files": []
-    }
+def get_max_uv_wavelength_from_txt(latest_file_path: str):
+    import pandas as pd
+    import os
+    # 文件检查
+    if not os.path.isfile(latest_file_path):
+        res = "ERROR: 指定的文件不存在"
+        return res
 
-    response = requests.post(url, headers=headers, json=data)
+    # 打开并读取最新文件
+    with open(latest_file_path, 'r') as file:
+        lines = file.readlines()
 
-    if response.status_code == 524:
-        print("Server is not responding. Please try again later. Maybe GPU was down in the container.")
-        return None
+    # 找到数据开始的行号
+    data_start_index = -1
+    for i, line in enumerate(lines):
+        if "Wavelength Scan Data Record" in line:
+            data_start_index = i + 2  # 数据从该行的下两行开始
+            break
 
-    try:
-        result = response.json()
-    except ValueError:
-        return [{"error": "Response is not in JSON format"}]
+    if data_start_index == -1:
+        res = "ERROR: 无法找到数据记录部分"
+        return res
 
-    useful_results = []
-    try:
-        answer = eval(result.get("answer", "[]"))
-        for item in answer:
-            metadata = item.get("metadata", {})
-            useful_info = {
-                "id": metadata.get("document_id"),
-                "title": item.get("title"),
-                "content": item.get("content"),
-                "metadata": None,
-                "embedding": None,
-                "score": metadata.get("score")
-            }
-            useful_results.append(useful_info)
-    except Exception as e:
-        return [{"error": f"Error processing result: {e}", "status": "TERMINATE"}]
-    if useful_results == []:
-        useful_results = "NULL"
-    return str(useful_results)
+    # 解析数据并构建表格
+    data = []
+    for line in lines[data_start_index:]:
+        parts = line.split()
+        if len(parts) == 7:  # 保证每行有7列数据
+            no, wavelength, abs_value, trans, energy, energy_100, energy_0 = parts
+            try:
+                data.append({
+                    'No': int(no),
+                    'Wavelength(nm)': float(wavelength),
+                    'Abs': float(abs_value),
+                    'Trans(%T)': float(trans),
+                    'Energy': float(energy),
+                    'Energy(100%T)': float(energy_100),
+                    'Energy(0%T)': float(energy_0)
+                })
+            except ValueError:
+                print(f"跳过无法解析的行: {line}")
+
+    if not data:
+        res = "ERROR: 未解析到任何有效数据"
+        return res
+
+    # 构建DataFrame
+    df = pd.DataFrame(data)
+
+    # 找到Abs值最大的行
+    max_abs_row = df.loc[df['Abs'].idxmax()]
+
+    # 获取最大Abs值对应的波长
+    max_abs_wavelength = max_abs_row['Wavelength(nm)']
+    res = f"本次实验的UV波长为: {max_abs_wavelength} nm"
+    print(res)
+    return res
+
+
+def get_max_pl_peak_from_txt(latest_file_path: str):
+    import pandas as pd
+    import os
+    # 文件检查
+    if not os.path.isfile(latest_file_path):
+        res = "ERROR: 指定的文件不存在"
+        return res
+
+    # 打开并读取最新文件
+    with open(latest_file_path, 'r') as file:
+        lines = file.readlines()
+
+    # 找到 'Data Points' 开始的行号
+    data_start_index = -1
+    for i, line in enumerate(lines):
+        if "Data Points" in line:
+            data_start_index = i + 1  # 数据从该行的下一行开始
+            break
+
+    if data_start_index == -1:
+        res = "ERROR: 无法找到数据记录部分"
+        return res
+
+    # 解析nm和Data数据
+    data = []
+    for line in lines[data_start_index:]:
+        parts = line.split()
+        if len(parts) == 2:  # 每行应该有2列数据，nm 和 Data
+            try:
+                nm = float(parts[0])
+                data_value = float(parts[1])
+                data.append({'nm': nm, 'Data': data_value})
+            except ValueError:
+                print(f"跳过无法解析的行: {line}")
+
+    if not data:
+        res = "ERROR: 未解析到任何有效数据"
+        return res
+
+    # 构建DataFrame
+    df = pd.DataFrame(data)
+
+    # 找到Data值最大的行
+    max_data_row = df.loc[df['Data'].idxmax()]
+
+    # 获取最大Data值对应的nm
+    max_data_nm = max_data_row['nm']
+
+    res = f"本次实验的PL峰位为: {max_data_nm} nm"
+    print(res)
+    return res