一种封装CMADaaS MUSIC Python接口的方法

>>> from nmc_met_io.retrieve_cmadaas import cmadaas_obs_by_time_and_id
>>> data = cmadaas_obs_by_time_and_id(
...     "20210419000000",
...     elements="Station_Id_C,Datetime,TEM",
...     sta_ids="54511"
... )
>>> data
  Station_Id_C   Datetime  TEM
0        54511 2021-04-19   16

>>> import pandas as pd
>>> from meteva import base as mb
>>> mb.io.set_io_config("~/.config/.nmcdev/config.ini")
>>> data = mb.io.read_stadata_from_cmadaas(
...     dataCode="SURF_CHN_MUL_HOR_N",
...     element="TEM",
...     time=pd.to_datetime("2021-04-19 00:00"),
...     station=pd.DataFrame({
...         "id": [54511]
...     })
... )
>>> data
   level                 time dtime  ...         lon        lat  SURF_CHN_MUL_HOR_N
0      0  2021-04-19 00:00:00     0  ...  116.466698  39.799999                16.0
[1 rows x 7 columns]

from nuwe_cmadaas.music import CMADaaSClient

client_config_path = "path/to/client/config/file"
user = "user name"
password = "user password"
server_id = "server id"

interface_id = "getSurfEleByTimeRange"

params = {
    "dataCode": "SURF_CHN_MUL_HOR",
    "elements": "Station_Id_d,Lat,Lon,Alti,Day,Hour,PRS_Sea,TEM,"
                "DPT,WIN_D_INST,WIN_S_INST,PRE_1h,PRE_6h,PRE_24h,PRS",
    "timeRange": "[20190817000000,20190817020000)",
    "orderby": "Station_ID_d:ASC",
    "limitCnt": "10",
}

client = CMADaaSClient(
    user=user,
    password=password,
    config_file=client_config_path
)
result = client.callAPI_to_array2D(interface_id, params)

def retrieve_obs_station(
        data_code: str = "SURF_CHN_MUL_HOR",
        elements: str = None,
        time: typing.Union[pd.Interval, pd.Timestamp, typing.List, pd.Timedelta] = None,
        station: typing.Union[str, typing.List, typing.Tuple] = None,
        region=None,
        station_level: typing.Union[str, typing.List[str]] = None,
        order: str = "Station_ID_d:asc",
        count: int = None,
        config_file: typing.Union[str, Path] = None,
        **kwargs,
) -> pd.DataFrame:
    pass

def retrieve_obs_grid(
        data_code: str,
        parameter: str = None,
        time: typing.Union[pd.Interval, pd.Timestamp, typing.List, pd.Timedelta] = None,
        station: typing.Union[str, typing.List, typing.Tuple] = None,
        region=None,
        station_level: typing.Union[str, typing.List[str]] = None,
        order: str = None,
        count: int = None,
        config_file: typing.Union[str, Path] = None,
        **kwargs,
) -> xr.DataArray:
    pass

def download_obs_file(
        data_code: str,
        elements: str = None,
        time: typing.Union[pd.Interval, pd.Timestamp, typing.List, pd.Timedelta] = None,
        station: typing.Union[str, typing.List, typing.Tuple] = None,
        region=None,
        station_level: typing.Union[str, typing.List[str]] = None,
        order: str = None,
        count: int = None,
        output_dir: str = "./",
        interface_data_type ="Surf",
        config_file: typing.Union[str, Path] = None,
        **kwargs,
):
    pass

if isinstance(time, pd.Interval):
    # 时间段
    interface_config["time"] = "TimeRange"
    pass
elif isinstance(time, pd.Timestamp):
    # 时间点
    interface_config["time"] = "Time"
    pass
elif isinstance(time, typing.List):
    # 时间列表
    interface_config["time"] = "Time"
    pass
elif isinstance(time, pd.Timedelta):
    # 时间间隔，用于获取最新时次
    interface_config["name"] = "getSurfLatestTime"
    pass

interface_config = {
    "name": "getSurfEle",
    "region": None,
    "time": None,
    "station": None,
}

interface_id = interface_config["name"]

region_part = interface_config["region"]
if region_part is not None:
    interface_id += "In" + region_part

condition_part = "And".join(filter(None, [
    interface_config["time"],
    interface_config["station"]
]))
if "level" in interface_config and interface_config["level"] is not None:
    condition_part = "And".join([condition_part, interface_config["level"]])

if len(condition_part) > 0:
    interface_id += "By" + condition_part

def _fix_interface_id(interface_id: str) -> str:
    mapper = {
        "getSurfEleByTimeRangeAndStaIdRange": "getSurfEleByTimeRangeAndStaIDRange"
    }
    return mapper.get(interface_id, interface_id)

def to_pandas(self) -> pd.DataFrame:
    df = pd.DataFrame(self.data, columns=self.element_names)
    return df

def to_xarray(self) -> xr.DataArray:
    coords = {}
    if len(self.lats) > 0:
        lats = self.lats
    else:
        lats = np.linspace(
            self.start_lat, self.end_lat, self.lat_count,
            endpoint=True
        )

    if len(self.lons) > 0:
        lons = self.lats
    else:
        lons = np.linspace(
            self.start_lon, self.end_lon, self.lon_count,
            endpoint=True
        )

    coords["latitude"] = xr.Variable(
        "latitude",
        lats,
        attrs={
            "units": "degrees_north",
            "standard_name": "latitude",
            "long_name": "latitude"
        },
    )
    coords["longitude"] = xr.Variable(
        "longitude",
        lons,
        attrs={
            "units": "degrees_east",
            "standard_name": "longitude",
            "long_name": "longitude"
        }
    )

    dims = ("latitude", "longitude")

    field = xr.DataArray(
        self.data,
        dims=dims,
        coords=coords,
        attrs={
            "units": self.units,
        },
        name=self.user_element_name,
    )

    return field

cmadaas:
  auth:
    user: user name
    password: user password
  server:
    music_server: music server ip
    music_port: music server port
    music_connTimeout: 3  # connection time out, seconds
    music_readTimeout: 3000  # read time out, seconds
    music_ServiceId: music service id

def _get_cmadaas_config_path(file_path: typing.Optional[str] = None):
    if file_path is None:
        if "NWPC_OPER_CONFIG" in os.environ:
            return os.environ["NWPC_OPER_CONFIG"]
        path = pathlib.Path(pathlib.Path.home(), ".config/nwpc-oper/config.yaml")
        return path
    else:
        return file_path

>>> import pandas as pd
>>> from nuwe_cmadaas.obs import retrieve_obs_station
>>> table = retrieve_obs_station(
...     "SURF_CHN_MUL_HOR",
...     time=pd.to_datetime("2021-04-19 00:00:00"),
...     elements="Station_Id_C,Datetime,TEM",
...     station="54511",
... )
>>> table
  Station_Id_C             Datetime TEM
0        54511  2021-04-19 00:00:00  16

from nuwe_cmadaas.obs import retrieve_obs_upper_air
retrieve_obs_upper_air(
    "UPAR_GLB_MUL_FTM",
    time=pd.Interval(
        pd.to_datetime("2021-01-01 00:00:00"),
        pd.to_datetime("2021-01-02 00:00:00"),
        closed="left"
    ),
    station=54511,
    level_type=("vertical", "pl"),
    level=(4096, 670),
)

from nuwe_cmadaas.model import retrieve_model_point
retrieve_model_point(
    "NAFP_ECMF_C1D_GLB_FOR",
    parameter="TEM",
    start_time=pd.to_datetime("2021-01-01 00:00:00"),
    forecast_time=("0h", "24h"),
    level_type=1,
    level=0,
    point=[(9.93, 116.28), (44.57, 129.6)]
)

from nuwe_cmadaas.model import retrieve_model_grid
retrieve_model_grid(
    "NAFP_ECMF_C1D_GLB_FOR",
    parameter="TEM",
    start_time=pd.to_datetime("2021-01-01 00:00:00"),
    forecast_time="24h",
    level_type=1,
    level=0,
    region={
        "type": "rect",
        "start_longitude": 70,
        "end_longitude": 140,
        "start_latitude": 55,
        "end_latitude": 15,
    }
)

from nuwe_cmadaas.model import download_model_file
download_model_file(
    "NAFP_FOR_FTM_GER_GLB",
    parameter="TEM",
    start_time=pd.to_datetime("2021-01-01 00:00:00"),
    forecast_time="12h",
    level_type="100",
    level=850,
    output_dir="./"
)