北斗电离层模型BDGIM广播系数

北斗全球电离层延迟修正模型BDGIM的广播系数是北斗卫星导航系统实时向用户播发的电离层延迟改正参数具有以下特点每个BDGIM预报包含9个球谐函数系数这些系数基于球谐函数展开模型能够精确描述全球电离层总电子含量TEC的空间分布特征。系统每2小时更新并播发一组新的系数每天共播发12组系数确保电离层改正信息的时效性。系数采用科学计数法表示格式为±X.XXXXXXXXXXXXe±XX提供15位有效数字的精度如2.787500000000e01表示27.875。系数通过北斗卫星的导航电文D2电文广播嵌入在BRD400DLR_S开头的广播星历文件中以 ION CXX CNVX标识符开头。1、系数下载来源武汉大学GNSS数据中心下载链接类似如下ftp://igs.gnsswhu.cn/pub/gps/data/daily/2025/brdc/BRD400DLR_S_20251890000_01D_MN.rnx.gz文件命名规范BRD400DLR_S_YYYYDOYHHMM_01D_MN.rnx.gzYYYY年份DOY年积日001-366HHMM起始时间解压处理需确保文件已从.gz格式解压为.rnx格式编码格式文件通常为ASCII或UTF-8编码2、数据批处理下载代码import os import ftplib import datetime import time import logging # 设置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) logger logging.getLogger(__name__) class GNSSDownloader: def __init__(self, ftp_hostigs.gnsswhu.cn): 初始化下载器 self.ftp_host ftp_host self.ftp None def connect_ftp(self): 连接到FTP服务器 try: self.ftp ftplib.FTP(self.ftp_host, timeout30) self.ftp.login() # 匿名登录 logger.info(f成功连接到FTP服务器: {self.ftp_host}) return True except Exception as e: logger.error(f连接FTP服务器失败: {e}) return False def disconnect_ftp(self): 断开FTP连接 if self.ftp: try: self.ftp.quit() except: self.ftp.close() logger.info(已断开FTP连接) def date_to_doy(self, date_obj): 将日期转换为年积日(DOY) year date_obj.year doy date_obj.timetuple().tm_yday return year, doy def generate_filename(self, year, doy): 生成文件名 # 格式: BRD400DLR_S_20220060000_01D_MN.rnx.gz doy_str f{doy:03d} filename fBRD400DLR_S_{year}{doy_str}0000_01D_MN.rnx.gz return filename def generate_remote_path(self, year, doy): 生成远程文件路径 # 示例: /pub/gps/data/daily/2022/brdc/ remote_path f/pub/gps/data/daily/{year}/brdc/ return remote_path def create_local_dir(self, year): 创建本地保存目录 - 只按年份创建目录 local_dir os.path.join(downloaded_data, str(year)) os.makedirs(local_dir, exist_okTrue) return local_dir def download_single_file(self, year, doy, retry_count3): 下载单个文件 # 生成文件名 filename self.generate_filename(year, doy) # 远程路径 remote_path self.generate_remote_path(year, doy) # 本地保存路径 - 直接放在年份目录下 local_dir self.create_local_dir(year) local_path os.path.join(local_dir, filename) # 检查文件是否已存在 if os.path.exists(local_path): file_size os.path.getsize(local_path) if file_size 1024: # 文件大于1KB才认为有效 logger.info(f文件已存在跳过: {filename} ({file_size} bytes)) return True, local_path # 尝试下载文件 for attempt in range(retry_count): try: logger.info(f正在下载: {filename} (尝试 {attempt 1}/{retry_count})) # 切换到远程目录 self.ftp.cwd(remote_path) # 获取文件大小 file_size None try: self.ftp.voidcmd(fSIZE {filename}) file_size int(self.ftp.sendcmd(fSIZE {filename}).split()[1]) except: pass # 下载文件 with open(local_path, wb) as f: self.ftp.retrbinary(fRETR {filename}, f.write) # 验证下载 if os.path.exists(local_path): actual_size os.path.getsize(local_path) if actual_size 0: if file_size and actual_size file_size: logger.info(f下载成功: {filename} ({actual_size} bytes, 完整)) else: logger.info(f下载成功: {filename} ({actual_size} bytes)) return True, local_path else: logger.warning(f下载的文件大小为0: {filename}) os.remove(local_path) except ftplib.error_perm as e: if 550 in str(e): # 文件不存在 logger.warning(f文件不存在: {filename}) return False, None logger.error(f权限错误: {filename} - {e}) except Exception as e: logger.error(f下载失败 {filename}: {e}) # 重试前等待 if attempt retry_count - 1: time.sleep(2) logger.error(f下载失败已达到最大重试次数: {filename}) return False, None def download_date_range(self, start_date, end_date): 下载指定日期范围的数据 Args: start_date: 开始日期格式 YYYY-MM-DD 或 datetime对象 end_date: 结束日期格式 YYYY-MM-DD 或 datetime对象 # 转换日期格式 if isinstance(start_date, str): start_date datetime.datetime.strptime(start_date, %Y-%m-%d) if isinstance(end_date, str): end_date datetime.datetime.strptime(end_date, %Y-%m-%d) # 连接到FTP服务器 if not self.connect_ftp(): logger.error(无法连接到FTP服务器程序退出) return try: # 遍历每一天 current_date start_date success_count 0 fail_count 0 missing_files [] logger.info(f开始下载 {start_date.date()} 到 {end_date.date()} 的数据) logger.info(f文件将保存在: downloaded_data/[年份]/ 目录下) while current_date end_date: year, doy self.date_to_doy(current_date) date_str current_date.strftime(%Y-%m-%d) logger.info(f处理: {date_str} (Year: {year}, DOY: {doy})) # 下载文件 success, local_path self.download_single_file(year, doy) if success: success_count 1 else: fail_count 1 missing_files.append(date_str) # 移动到下一天 current_date datetime.timedelta(days1) # 短暂延迟避免请求过快 time.sleep(0.5) # 下载结果统计 logger.info( * 50) logger.info(f下载完成!) logger.info(f成功: {success_count} 个文件) logger.info(f失败: {fail_count} 个文件) logger.info(f文件保存位置: downloaded_data/ 目录下) if missing_files: logger.info(缺失的文件日期:) for date in missing_files: logger.info(f - {date}) except KeyboardInterrupt: logger.info(用户中断下载) except Exception as e: logger.error(f下载过程中发生错误: {e}) finally: self.disconnect_ftp() def download_specific_years(self, years): 下载特定年份的数据 Args: years: 年份列表如 [2022, 2023, 2024] for year in years: logger.info(f开始下载 {year} 年的数据) # 设置该年的开始和结束日期 start_date datetime.datetime(year, 1, 1) # 判断是否为闰年 if (year % 4 0 and year % 100 ! 0) or (year % 400 0): end_date datetime.datetime(year, 12, 31) else: end_date datetime.datetime(year, 12, 31) # 下载该年份的数据 self.download_date_range(start_date, end_date) def download_specific_dates(self, date_list): 下载特定日期的数据 Args: date_list: 日期列表格式 [YYYY-MM-DD, YYYY-MM-DD, ...] if not self.connect_ftp(): logger.error(无法连接到FTP服务器程序退出) return try: success_count 0 fail_count 0 for date_str in date_list: try: date_obj datetime.datetime.strptime(date_str, %Y-%m-%d) year, doy self.date_to_doy(date_obj) logger.info(f处理: {date_str} (Year: {year}, DOY: {doy})) success, _ self.download_single_file(year, doy) if success: success_count 1 else: fail_count 1 except ValueError: logger.error(f日期格式错误: {date_str}) fail_count 1 # 短暂延迟 time.sleep(0.5) logger.info(f特定日期下载完成: 成功 {success_count}, 失败 {fail_count}) finally: self.disconnect_ftp() def check_existing_files(self, yearNone): 检查已下载的文件 Args: year: 要检查的年份如果为None则检查所有年份 base_dir downloaded_data if not os.path.exists(base_dir): logger.info(f下载目录不存在: {base_dir}) return if year: # 检查特定年份 year_dir os.path.join(base_dir, str(year)) if os.path.exists(year_dir): files os.listdir(year_dir) gnss_files [f for f in files if f.startswith(BRD4)] logger.info(f{year}年已下载 {len(gnss_files)} 个文件:) for file in sorted(gnss_files)[:10]: # 只显示前10个文件 logger.info(f - {file}) if len(gnss_files) 10: logger.info(f ... 还有 {len(gnss_files) - 10} 个文件) else: logger.info(f{year}年目录不存在) else: # 检查所有年份 years [d for d in os.listdir(base_dir) if os.path.isdir(os.path.join(base_dir, d))] logger.info(已下载的年份:) for year_dir in sorted(years): year_path os.path.join(base_dir, year_dir) if os.path.isdir(year_path): files os.listdir(year_path) gnss_files [f for f in files if f.startswith(BRD4)] logger.info(f {year_dir}: {len(gnss_files)} 个文件) def main(): 主函数 - 下载数据 # 创建下载器实例 downloader GNSSDownloader() # 设置下载日期范围 start_date 2025-01-01 end_date 2025-12-31 # 开始下载 downloader.download_date_range(start_date, end_date) # 检查已下载的文件 downloader.check_existing_files() if __name__ __main__: print(北斗广播星历文件下载工具) main()3、提取广播系数标识符定位系数块以 ION CXX CNVX开头CXX中的XX为卫星编号如C29、C19等CNVX表示电离层改正参数块结构每个块包含3行数据第1行时间戳 前3个系数第2行中间4个系数第3行最后2个系数科学计数法所有系数均为科学计数法格式格式±X.XXXXXXXXXXXXe±XX示例2.787500000000e01无间隔系数注意系数间可能没有空格正确解析2.787500000000e01-4.500000000000e00应分割为2.787500000000e01和-4.500000000000e00系数数量每个时间点固定为9个系数4、汇总处理结果汇总后发现有些历元竟然有不同的广播系数播发间隔也是五花八门不是固定的时间间隔查看原文件确实如此不太懂是什么原因求解答。