Convert NetCDF To Zarr - Example 2

# 1. Define your connection info
host = "eodata-bec.icm.csic.es"
port = 25288 
username = ""
password = ""

import paramiko

# 2. Initialize the SSH client
ssh = paramiko.SSHClient()

# Automatically add the server's host key (prevents "unknown host" errors)
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Connect to the server
ssh.connect(hostname=host, port=port, username=username, password=password)

# Open the SFTP session on the SSH connection
sftp = ssh.open_sftp()
print(f"Successfully connected to {host} via SFTP!\n")

# 3. List the files in the current directory
print("Files currently on the server:")
server_files = sftp.listdir() # listdir() returns a list of file names
for file_name in server_files:
    print(f" - {file_name}")
print("\n")

root_path = 'OCEAN/SSS/SMOS/SouthernOcean/v1.0/L3/9day/'

import os

for folder in sorted(sftp.listdir(root_path)):
    if folder.isalnum() and folder != '2011':
        os.mkdir(f'./data/{folder}/')
        print('Processing year :' + folder)
        print('Downloading : ', len(sorted(sftp.listdir(root_path + folder))))
        for f in sorted(sftp.listdir(root_path + folder)):
            sftp.get(root_path + folder + '/' + f, f'./data/{folder}/{f}')

import xarray as xr
xr.open_dataset('./data/2011/BEC_SSS___SMOS__SO__L3__B_20110208T120000_25km__9d_REP_v100.nc')

import re
import numpy as np
import xarray as xr
from typing import Iterable, List, Sequence
from datetime import datetime, timedelta
from pathlib import Path

FILENAME_RE = re.compile(r"_(\d{8})T120000_")


input_root = Path('./data/')

files = sorted(input_root.glob("[0-9][0-9][0-9][0-9]/*.nc"))
len(files)

def file_center_datetime(file_path: Path) -> datetime:
    match = FILENAME_RE.search(file_path.name)
    if not match:
        raise ValueError(f"Cannot parse date from filename: {file_path.name}")
    return datetime.strptime(match.group(1), "%Y%m%d")


def build_time_bounds(files: Iterable[Path]) -> np.ndarray:
    starts: List[np.datetime64] = []
    ends: List[np.datetime64] = []
    for f in files:
        center = file_center_datetime(f)
        start = center
        end = center + timedelta(days=8, hours=23, minutes=59, seconds=59)
        starts.append(np.datetime64(start, "ns"))
        ends.append(np.datetime64(end, "ns"))
    return np.stack([starts, ends], axis=1)


def open_dataset(
    files: Sequence[Path],
    t_chunk: int = 10,
    y_chunk: int = 720,
    x_chunk: int = 720,
    parallel: bool = True,
    engine: str = 'netcdf4',
) -> xr.Dataset:
    ds = xr.open_mfdataset(
        [str(f) for f in files],
        combine="nested",
        concat_dim="time",
        data_vars="minimal",
        coords="minimal",
        compat="override",
        parallel=parallel,
        engine=engine,
        chunks={"time": 1, "y": 720, "x": 720},
    ).sortby("time")

    ds = ds.chunk({"time": t_chunk, "y": y_chunk, "x": x_chunk})

    time_bnds = build_time_bounds(files)
    ds["time_bnds"] = xr.DataArray(
        time_bnds,
        dims=("time", "nv"),
        attrs={
            "long_name": "time bounds",
        },
    )
    ds["time"].attrs["bounds"] = "time_bnds"
    return ds

ds = open_dataset(files=files)
ds.attrs['time_coverage_end'] = ds.time.values[-1]
ds

outputdir = 'sofresh.zarr'
ds.to_zarr(
    outputdir,
    mode='w',
    zarr_format = 3,
    consolidated=True

)

ds = xr.open_zarr(outputdir)
ds

rnd_file = np.random.choice(files[:1000], size=1)[0]
orig = xr.open_dataset(rnd_file)
orig

assert np.isclose(ds.sel(time=orig.time.values[0]).sss.values, orig.sss.values, equal_nan=True).all()
assert np.isclose(ds.sel(time=orig.time.values[0]).uncertainty_sss.values, orig.uncertainty_sss.values, equal_nan=True).all()