计算sif值，上传到地调院服务器；

.gitignore

@@ -2,6 +2,7 @@
 /.idea
 test_data
 tmp
+result
 
 
 # Byte-compiled / optimized / DLL files

config.ini

@@ -1,9 +1,9 @@
 [FTP]
-host = 172.16.0.73
+host = 183.207.7.90
-port = 22
+port = 65521
-user = ftpuser
+user = sif
-password = 123
+password = groundiot@2025
-target_dir = /home/ftpuser/
+target_dir = /eco_monitoring/ground_iot/sif
 [monitor]
 WATCH_DIR = D:\PycharmProjects\sif_data_parse\test_data
 cal_dir = D:\PycharmProjects\sif_data_parse\test_data\cal

main.py

@@ -12,6 +12,10 @@ from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
 from datetime import datetime
 
+import ssl
+from ftplib import FTP_TLS, error_perm
+
+
 
 def parse_sif_csv(_file_path):
     _metadata = {}
@@ -140,24 +144,89 @@ class CSVFileHandler(FileSystemEventHandler):
             return
         if event.src_path.lower().endswith('.csv'):
             file_path = os.path.abspath(event.src_path)
-            print(f"发现CSV文件: {file_path}")
+            print(f"发现CSV文件: {file_path}----------------------------------------------------------------------------")
 
             # 选择定标文件夹
             cal_dir = self.ftp_config['monitor']['cal_dir']
-            a=1
             if "towersif20" in file_path:
                 cal_dir = os.path.join(cal_dir,"20")
             elif "towersif21" in file_path:
                 cal_dir = os.path.join(cal_dir,"21")
 
             time.sleep(0.1)  # 文件一出现就处理文件，偶发permission deny，所以等待100ms
-            _ = self.process_csv(file_path, cal_dir)
 
-            # 为csv添加有效性字段
+            rad_folder_tmp, sif_folder, quality_level = self.rad_conversion(file_path, cal_dir)
+            _ = self.compute_sif(file_path, rad_folder_tmp, sif_folder)
             for i in _:
-                self.add_validity_column_to_file(i)
+                self.add_validity_column_to_file(i, quality_level)
 
-            self.send_via_sftp(_)
+            self.send_via_ftps(_)
+
+    def send_via_ftps(self, file_paths, max_retries=3, retry_delay=5):
+        retries = 0
+        ftps = None
+
+        while retries < max_retries:
+            try:
+                print("正在尝试连接 FTPS 服务器...")
+
+                # 建立 FTPS 连接
+                ftps = FTP_TLS()
+                # 忽略自签名证书（你的服务器证书是 self-signed）
+                ftps.context = ssl._create_unverified_context()
+                ftps.connect(
+                    host=self.ftp_config['FTP']['host'],
+                    port=int(self.ftp_config['FTP'].get('port', 21)),  # 默认 21, 你的可能是 65521
+                    timeout=30
+                )
+
+                # 登录
+                ftps.login(
+                    user=self.ftp_config['FTP']['user'],
+                    passwd=self.ftp_config['FTP']['password']
+                )
+                print("FTPS 连接成功，准备上传文件...")
+
+                # 切换到安全数据通道
+                ftps.prot_p()
+
+                # 检查并切换到目标目录
+                remote_dir = self.ftp_config['FTP'].get('target_dir', '.')
+                try:
+                    ftps.cwd(remote_dir)
+                except error_perm:
+                    print(f"远程目录不存在，尝试创建: {remote_dir}")
+                    ftps.mkd(remote_dir)
+                    ftps.cwd(remote_dir)
+
+                # 上传多个文件
+                success_count = 0
+                for file_path in file_paths:
+                    try:
+                        filename = os.path.basename(file_path)
+                        with open(file_path, "rb") as f:
+                            ftps.storbinary(f"STOR {filename}", f)
+                        print(f"✅ 文件上传成功: {filename}")
+                        success_count += 1
+                    except Exception as e:
+                        print(f"❌ 文件上传失败 {file_path}: {e}")
+
+                return success_count == len(file_paths)  # 全部成功返回 True，否则 False
+
+            except Exception as e:
+                retries += 1
+                print(f"❌ FTPS 连接失败（尝试 {retries}/{max_retries}）: {e}")
+                if retries < max_retries:
+                    time.sleep(retry_delay)
+            finally:
+                if ftps:
+                    try:
+                        ftps.quit()
+                    except Exception:
+                        ftps.close()
+
+        print(f"❌ 上传失败（已达最大重试次数 {max_retries}）")
+        return False
 
     def send_via_sftp(self, file_paths, max_retries=3, retry_delay=5):
         retries = 0
@@ -228,7 +297,7 @@ class CSVFileHandler(FileSystemEventHandler):
         print(f"❌ 上传失败（已达最大重试次数 {max_retries}）")
         return False
 
-    def process_csv(self, input_csv, input_cal):
+    def rad_conversion(self, input_csv, input_cal):
         # 提取文件夹路径
         folder_path = os.path.dirname(input_csv)
         base_name = os.path.basename(input_csv)  # 获取文件名（含扩展名）
@@ -238,27 +307,44 @@ class CSVFileHandler(FileSystemEventHandler):
         formatted_date = today.strftime("%Y_%m_%d")
         new_name = f"{formatted_date}_{parts[1]}{ext}"  # 组合新文件名
 
-        # tmp_folder = os.path.join(os.path.dirname(ftp_config['monitor']['WATCH_DIR']), "tmp")
+        result_folder = os.path.join(Path(ftp_config['monitor']['WATCH_DIR']).parent, "result")
-        tmp_folder = os.path.join(Path(ftp_config['monitor']['WATCH_DIR']).parent, "tmp")
+        if not os.path.exists(result_folder):
-        if not os.path.exists(tmp_folder):
+            os.makedirs(result_folder)
-            os.makedirs(tmp_folder)
+            print(f"文件夹已创建: {result_folder}")
-            print(f"文件夹已创建: {tmp_folder}")
         else:
-            print(f"文件夹已存在: {tmp_folder}")
+            print(f"文件夹已存在: {result_folder}")
 
-        rad_folder = os.path.join(tmp_folder, "rad")
+        def get_last_two_levels(path):
-        if os.path.exists(rad_folder):
+            path_obj = Path(path)
-            shutil.rmtree(rad_folder)
+            parts = path_obj.parts
+            # 获取最后两层
+            if len(parts) >= 2:
+                return Path(parts[-2]) / parts[-1]
+            else:
+                return path_obj  # 如果路径少于两层，返回原路径
+
+        # folder_name = os.path.basename(folder_path)
+        folder_name = get_last_two_levels(folder_path)
+
+        rad_folder = os.path.join(result_folder, folder_name,  "rad")
+        if not os.path.exists(rad_folder):
             os.makedirs(rad_folder)
+            print(f"文件夹已创建: {rad_folder}")
+        else:
+            print(f"文件夹已存在: {rad_folder}")
 
-        sif_folder = os.path.join(tmp_folder, "sif")
+        sif_folder = os.path.join(result_folder, folder_name,  "sif")
         if not os.path.exists(sif_folder):
             os.makedirs(sif_folder)
             print(f"文件夹已创建: {sif_folder}")
         else:
             print(f"文件夹已存在: {sif_folder}")
 
-        rad_path = os.path.join(rad_folder, new_name)
+        rad_folder_tmp = os.path.join(result_folder, folder_name,  "rad_tmp")
+        if os.path.exists(rad_folder_tmp):
+            shutil.rmtree(rad_folder_tmp)
+        os.makedirs(rad_folder_tmp)
+        rad_path = os.path.join(rad_folder_tmp, new_name)
 
         metadata, wavelengths, spectra_data = parse_sif_csv(input_csv)
 
@@ -272,6 +358,72 @@ class CSVFileHandler(FileSystemEventHandler):
             spectra_data[i]['RAD'] = spectra_data[i]['DN'] * data_gain_adjust
 
         write_file(input_csv, rad_path, spectra_data)
+        shutil.copy(rad_path, rad_folder)
+
+        # 计算有效性
+        rad_file_name = os.path.basename(rad_path)
+        quality_level = self.compute_quality_level(rad_file_name, rad_folder)
+
+        return rad_folder_tmp, sif_folder, quality_level
+
+    def compute_quality_level(self, file_name, folder_path, center=753, window=3):
+        all_files = [f for f in os.listdir(folder_path) if f.endswith('.csv')]
+
+        # 处理文件夹为空或只有给定文件的情况
+        if not all_files or file_name not in all_files:
+            return 2
+
+        def parse_time_from_filename(filename):
+            # 去掉.csv后缀，然后用下划线分割
+            time_str = filename.rsplit('.', 1)[0]
+            try:
+                # 将字符串转换为datetime对象
+                return datetime.strptime(time_str, '%Y_%m_%d_%H_%M_%S')
+            except ValueError:
+                # 如果文件名不符合格式，返回一个遥远的未来时间，确保它被排到最后
+                return datetime.max
+
+        sorted_files = sorted(all_files, key=parse_time_from_filename)
+
+        index_of_given = sorted_files.index(file_name)
+
+        if index_of_given==0:
+            return 2
+
+        current_file = os.path.join(folder_path, file_name)
+        metadata1, wavelengths1, spectra_data1 = parse_sif_csv(current_file)
+
+        pre_file = os.path.join(folder_path, sorted_files[index_of_given-1])
+        metadata2, wavelengths2, spectra_data2 = parse_sif_csv(pre_file)
+
+        lower_bound = center - window
+        upper_bound = center + window
+
+        indices = np.where((wavelengths2 >= lower_bound) & (wavelengths2 <= upper_bound))[0]
+
+        pos = 1  # 以csv文件中那个位置计算可信度
+        a1 = spectra_data1[pos]['DN'][indices].mean()
+        a2 = spectra_data2[pos]['DN'][indices].mean()
+
+        quality_number = abs(a2 - a1) / a1 * 100
+
+        if quality_number <= 2:
+            return 0
+        elif quality_number <= 5:
+            return 1
+        elif quality_number <= 10:
+            return 2
+        else:
+            return 3
+
+    def compute_sif(self, input_csv_dn, rad_folder, sif_folder):
+        folder_path = os.path.dirname(input_csv_dn)
+        base_name = os.path.basename(input_csv_dn)  # 获取文件名（含扩展名）
+        name_part, ext = os.path.splitext(base_name)  # 拆分文件名和扩展名
+        parts = name_part.split('_', 1)  # 在第一个 _ 处分割
+        today = datetime.now()
+        formatted_date = today.strftime("%Y_%m_%d")
+        new_name = f"{formatted_date}_{parts[1]}{ext}"  # 组合新文件名
 
         # 调用丰算法
         if os.name == "nt": # Windows
@@ -305,7 +457,7 @@ class CSVFileHandler(FileSystemEventHandler):
 
         return output_path_3fld, output_path_sfld, output_path_sfm
 
-    def add_validity_column_to_file(self, file_path):
+    def add_validity_column_to_file(self, file_path, quality_level):
         # 创建临时文件
         temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False, newline='')
 
@@ -322,7 +474,7 @@ class CSVFileHandler(FileSystemEventHandler):
 
                 # 添加validity列
                 rows[0].append('validity')
-                rows[1].append('1')
+                rows[1].append(quality_level)
 
                 # 写入临时文件
                 writer.writerows(rows)