Python学习|RS Python入门02:列表、字典与流程控制:像元集合的编队【四】for循环

for item in sequence:    process(item)  # 对每个item执行操作

for i in range(start, stop, step):    process_with_index(i)  # 使用索引i

bands = ["Blue", "Green", "Red", "NIR"]for band in bands:    print(f"处理波段: {band}")

coordinates = (120.5, 31.2, 115.8, 39.9)for coord in coordinates:    print(f"坐标值: {coord}")

for char in "NDVI":    print(f"字符: {char}")

metadata = {"sensor": "Landsat8", "date": "2023-06-15", "cloud_cover": 12.5}for key in metadata:    print(f"键: {key}")for value in metadata.values():    print(f"值: {value}")for key, value in metadata.items():    print(f"{key}: {value}")

import osimage_dir = "/data/landsat/time_series/"image_files = [f for f in os.listdir(image_dir) if f.endswith(".tif")]processed_count = 0for filename in image_files:    file_path = os.path.join(image_dir, filename)    print(f"正在处理: {filename}")    result = process_single_image(file_path)    processed_count += 1    print(f"成功处理: {filename}")    print(f"处理失败 {filename}: {str(e)}")print(f"处理完成，共处理 {processed_count} 景影像")

def calculate_spectral_indices(image_data, band_names):    indices = {}    for band_name in band_names:        if band_name.startswith("B"):                print(f"预处理波段: {band_name}")    if "Red" in image_data and "NIR" in image_data:        red = image_data["Red"]        nir = image_data["NIR"]        ndvi = (nir - red) / (nir + red + 1e-10)        indices["NDVI"] = ndvi        print("计算NDVI完成")    spectral_indices_to_calculate = [        ("NDWI", ["Green", "NIR"]),        ("EVI", ["Blue", "Red", "NIR"]),        ("SAVI", ["Red", "NIR"])    ]    for index_name, required_bands in spectral_indices_to_calculate:        if all(band in image_data for band in required_bands):            index_value = calculate_specific_index(index_name, image_data)            indices[index_name] = index_value            print(f"计算{index_name}完成")    return indices

def analyze_time_series(image_series, dates):    results = {        "trend": None,        "seasonality": [],        "anomalies": []    }    for i, (image, date) in enumerate(zip(image_series, dates)):        print(f"分析日期: {date}")        current_stats = calculate_image_statistics(image)        if i > 0:            previous_stats = calculate_image_statistics(image_series[i-1])            change = current_stats["mean"] - previous_stats["mean"]            if abs(change) > previous_stats["std"] * 2:                results["anomalies"].append({                    "date": date,                    "change": change,                    "type": "significant_change"                })                print(f"检测到异常变化: {date}")    if len(image_series) >= 3:        results["trend"] = calculate_trend(image_series)        print("趋势分析完成")    return results

rows = 10cols = 10print("网格坐标:")for row in range(rows):    for col in range(cols):        print(f"({row}, {col})", end=" ")    print()

print("反向遍历波段:")for i in range(10, 0, -1):    print(f"波段 B{i}")

print("每隔3个像元采样:")width = 100for x in range(0, width, 3):    print(f"采样列: {x}")

bands = ["Blue", "Green", "Red", "NIR", "SWIR1", "SWIR2"]print("波段列表:")for index, band in enumerate(bands):    print(f"索引 {index}: {band}")

print("\n从1开始计数:")for index, band in enumerate(bands, start=1):    print(f"波段 {index}: {band}")

def normalize_bands(band_data_list):    normalized_bands = []    for i, band_data in enumerate(band_data_list):        print(f"正在归一化波段 {i+1}")        min_val = band_data.min()        max_val = band_data.max()        normalized = (band_data - min_val) / (max_val - min_val)        normalized_bands.append(normalized)        print(f"波段 {i+1}: 范围 [{min_val:.3f}, {max_val:.3f}]")    return normalized_bands

def process_multispectral_image(image_bands, metadata):results = {}for idx, (band_name, band_data) in enumerate(image_bands.items()):    print(f"处理第 {idx+1}/{len(image_bands)} 个波段: {band_name}")    band_metadata = metadata.get("bands", {}).get(band_name, {})    if band_metadata.get("requires_radiometric_correction", False):        print(f"  对 {band_name} 进行辐射校正")        band_data = apply_radiometric_correction(band_data, band_metadata)    if band_metadata.get("requires_atmospheric_correction", False):        print(f"  对 {band_name} 进行大气校正")        band_data = apply_atmospheric_correction(band_data, band_metadata)    results[band_name] = band_datareturn results

band_names = ["B2", "B3", "B4", "B8"]wavelengths = [490, 560, 665, 842]resolutions = [10, 10, 10, 10]print("波段详细信息:")for name, wavelength, resolution in zip(band_names, wavelengths, resolutions):    print(f"{name}: {wavelength}nm, {resolution}m")

from itertools import zip_longestshort_list = ["B1", "B2", "B3"]long_list = [443, 490, 560, 665, 842]print("\n使用zip:")for a, b in zip(short_list, long_list):    print(f"{a}: {b}")print("\n使用zip_longest:")for a, b in zip_longest(short_list, long_list, fillvalue="N/A"):    print(f"{a}: {b}")

def calculate_band_ratios(band_data_dict):    ratios = {}    band_names = list(band_data_dict.keys())    for i, band1 in enumerate(band_names):        for j, band2 in enumerate(band_names):            if i >= j:                continue            ratio_name = f"{band1}_{band2}_ratio"            data1 = band_data_dict[band1]            data2 = band_data_dict[band2]            with np.errstate(divide='ignore', invalid='ignore'):                ratio = np.where(data2 != 0, data1 / data2, 0)            ratios[ratio_name] = ratio    return ratiosdef align_time_series(images, dates, quality_scores):    aligned_data = []    for image, date, quality in zip(images, dates, quality_scores):        if quality >= 0.8:            aligned_data.append({                "date": date,                "image": image,                "quality": quality            })        else:            print(f"跳过低质量影像: {date}")    return aligned_data

squares = []for i in range(1, 11):    squares.append(i ** 2)

squares = [i ** 2 for i in range(1, 11)]

years = [2020, 2021, 2022, 2023]months = ["01", "06"]# 生成所有年月组合的文件名file_names = [f"LC08_{year}{month}01.tif"              for year in years              for month in months]print("生成的文件名:", file_names[:5])

all_bands = ["B1", "B2", "B3", "B4", "B5", "B6", "B7", "B8", "B8A", "B9", "B10", "B11", "B12"]visible_bands = [band for band in all_bandsif band in ["B2", "B3", "B4"]]print("可见光波段:", visible_bands)

images = [    {"name": "img1", "cloud_cover": 5.2, "resolution": 30},    {"name": "img2", "cloud_cover": 35.1, "resolution": 30},    {"name": "img3", "cloud_cover": 12.3, "resolution": 10},    {"name": "img4", "cloud_cover": 8.7, "resolution": 30}]# 筛选低云量、高分辨率的影像selected_images = [    img for img in images    if img["cloud_cover"] < 20 and img["resolution"] <= 30]print("选中的影像:", [img["name"] for img in selected_images])

rows, cols = 3, 4

grid = []for i in range(rows):    row = []    for j in range(cols):        row.append(i * cols + j)    grid.append(row)

grid = [[i * cols + j for j in range(cols)] for i in range(rows)]print("栅格数据:")for row in grid:    print(row)

def extract_all_pixel_values(raster_data):    all_values = [        raster_data[b, r, c]        for b in range(raster_data.shape[0])        for r in range(raster_data.shape[1])        for c in range(raster_data.shape[2])    ]    return all_values

band_codes = ["B01", "B02", "B03", "B04"]wavelengths = [443, 490, 560, 665]band_to_wavelength = {band: wavelength                      for band, wavelength in zip(band_codes, wavelengths)}print("波段-波长映射:", band_to_wavelength)

bands = ["Coastal_Aerosol", "Blue", "Green", "Red", "NIR"]band_dict = {band: f"Band_{i+1}" for i, band in enumerate(bands)}print("波段字典:", band_dict)

sensor_records = [    {"sensor": "Landsat8", "date": "2023-01-01"},    {"sensor": "Sentinel2", "date": "2023-01-02"},    {"sensor": "Landsat8", "date": "2023-01-03"},    {"sensor": "MODIS", "date": "2023-01-04"},    {"sensor": "Sentinel2", "date": "2023-01-05"}]unique_sensors = {record["sensor"] for record in sensor_records}print("唯一传感器类型:", unique_sensors)

band_sets = [    {"B1", "B2", "B3", "B4"},    {"B2", "B3", "B4", "B5"},    {"B3", "B4", "B5", "B8"},    {"B4", "B5", "B8", "B11"}]common_bands = set.intersection(*band_sets)print("共同波段:", common_bands)

from itertools import product, combinations, permutations# product 用于生成多个序列的笛卡尔积，常用于枚举所有可能的处理参数组合。resolutions = [10, 20, 30]compression_methods = ["LZW", "DEFLATE", "JPEG"]print("所有可能的处理组合:")for res, comp in product(resolutions, compression_methods):    print(f"分辨率: {res}m, 压缩: {comp}")

bands = ["B1", "B2", "B3", "B4"]print("\n所有可能的波段组合（3个一组）:")for combo in combinations(bands, 3):    print(combo)

print("\n所有可能的波段比值:")for band1, band2 in combinations(bands, 2):    print(f"{band1}/{band2}")

def process_large_image_files(file_list):for file_path in file_list:print(f"正在读取: {file_path}")    for chunk in read_image_in_chunks(file_path):        processed_chunk = process_chunk(chunk)        yield processed_chunk

large_files = ["big_image1.tif", "big_image2.tif", "big_image3.tif"]for processed_chunk in process_large_image_files(large_files):    save_chunk(processed_chunk)

ndvi_values = (calculate_ndvi(pixel) for pixel in large_pixel_stream)for ndvi in ndvi_values:    if ndvi > 0.6:        classify_vegetation(ndvi)

import asyncioasync def process_multiple_images_async(image_paths):    tasks = []    for path in image_paths:        task = asyncio.create_task(process_single_image_async(path))        tasks.append(task)    results = await asyncio.gather(*tasks, return_exceptions=True)    successful = 0    for i, result in enumerate(results):        if isinstance(result, Exception):            print(f"处理失败 {image_paths[i]}: {result}")        else:            successful += 1    print(f"成功处理 {successful}/{len(image_paths)} 个文件")    return resultsasync def process_single_image_async(file_path):    await asyncio.sleep(0.1)    return {"file": file_path, "status": "processed"}

def process_bands_slow(band_data):    results = []    for band in band_data:        iflen(band_data) > 5:            result = expensive_computation(band)        else:            result = simple_computation(band)        results.append(result)    return results

def process_bands_fast(band_data):    results = []    band_count = len(band_data)    if band_count > 5:        process_func = expensive_computation    else:        process_func = simple_computation    for band in band_data:        results.append(process_func(band))    return results

def calculate_statistics_fast(data):sum_func = sumlen_func = lenappend_func = results.appendresults = []for subset in data:    mean_val = sum_func(subset) / len_func(subset)    append_func(mean_val)return results

import numpy as npdef normalize_bands_loop(band_list):    normalized = []    for band in band_list:        min_val = np.min(band)        max_val = np.max(band)        norm_band = (band - min_val) / (max_val - min_val)        normalized.append(norm_band)    return normalizeddef normalize_bands_vectorized(band_list):    bands_array = np.array(band_list)    min_vals = np.min(bands_array, axis=(1, 2), keepdims=True)    max_vals = np.max(bands_array, axis=(1, 2), keepdims=True)    normalized = (bands_array - min_vals) / (max_vals - min_vals)    return list(normalized)

import timetest_bands = [np.random.rand(1000, 1000) for _ in range(10)]start = time.time()result_loop = normalize_bands_loop(test_bands)print(f"循环版本耗时: {time.time() - start:.4f}秒")start = time.time()result_vec = normalize_bands_vectorized(test_bands)print(f"向量化版本耗时: {time.time() - start:.4f}秒")