mosdac_preprocessing.py

import os
import h5py
import numpy as np
from PIL import Image
from datetime import datetime

# Base output directory
output_dir = "outputtest"

# Channels to process
CHANNELS = {
    "IMG_VIS": ("IMG_VIS_ALBEDO", "albedo"),         # in %
    "IMG_SWIR": ("IMG_SWIR_ALBEDO", "albedo"),       # in %
    "IMG_MIR": ("IMG_MIR_RADIANCE", "radiance"),     # mW.cm-2.sr-1.micron-1
    "IMG_WV": ("IMG_WV_TEMP", "temperature"),        # in Kelvin
    "IMG_TIR1": ("IMG_TIR1_TEMP", "temperature"),    # in Kelvin
    "IMG_TIR2": ("IMG_TIR2_TEMP", "temperature"),    # in Kelvin
}

# Path to folder containing .h5 files
h5_folder = r"C:\Users\HP\Desktop\MOSDAC" 
#h5_folder = r"C:\Users\HP\Desktop\PROCESSEDMOSDAC\testingtemp"
def convert_dn(dn_array, lookup):
    return lookup[dn_array]

def normalize_and_save(img, save_path):
    img = np.nan_to_num(img)
    img_min, img_max = np.min(img), np.max(img)
    norm_img = ((img - img_min) / (img_max - img_min + 1e-5) * 255).astype(np.uint8)
    Image.fromarray(norm_img).save(save_path)

def process_file(file_path):
    file_name = os.path.basename(file_path)
    # Extract date and time
    parts = file_name.split("_")
    date_str = parts[1]  # e.g., "01JUN2025"
    time_str = parts[2]  # e.g., "0300"

    # Convert to YYYYMMDD and HHMM format
    date_dt = datetime.strptime(date_str, "%d%b%Y")
    date_fmt = date_dt.strftime("%Y%m%d")
    time_fmt = time_str

    output_path = os.path.join(output_dir, date_fmt, time_fmt)
    os.makedirs(output_path, exist_ok=True)

    with h5py.File(file_path, "r") as f:
        for ch, (lookup_key, label_type) in CHANNELS.items():
            if ch not in f or lookup_key not in f:
                print(f"Skipping missing channel: {ch}")
                continue

            dn = f[ch][0]
            lookup = f[lookup_key][:]
            processed = convert_dn(dn, lookup)
            save_path = os.path.join(output_path, f"{ch}.png")
            normalize_and_save(processed, save_path)
            print(f"Saved: {save_path}")

# Main loop
for filename in os.listdir(h5_folder):
    if filename.endswith(".h5"):
        process_file(os.path.join(h5_folder, filename))