from typing import Optional, Union, Dict, Any, Tuple
import io
import logging
import numpy as np
import pandas as pds
from speasy.core.codecs.bundled_codecs.hapi.reader import _extract_headers, _load_hapi
from speasy.core.codecs.bundled_codecs.hapi.hapi_file import HapiFile
from speasy.core.codecs.codec_interface import Buffer
from speasy.core.any_files import any_loc_open
log = logging.getLogger(__name__)
def _extract_data_csv(file: io.IOBase, headers: Dict[str, Any]) -> pds.DataFrame:
data = io.BytesIO(file.read())
return pds.read_csv(data, comment='#', sep=',', header=None, skiprows=0, parse_dates=[0], index_col=0)
[docs]
def load_hapi_csv(file: Union[Buffer, str, io.IOBase]) -> Optional[HapiFile]:
data, headers = _load_hapi(file, _extract_data_csv)
hapi_csv_file = HapiFile()
if data is not None and headers is not None:
time_header = headers["parameters"][0]
assert time_header["type"] == "isotime"
hapi_csv_file.create_parameter(data.index.to_numpy(dtype="datetime64[ns]"),
meta=time_header)
column_offset = 0
for i, param_meta in enumerate(headers["parameters"][1:]):
shape = param_meta.get("size", [1])
flatten_shape = np.prod(shape)
hapi_csv_file.create_parameter(data.iloc[:, column_offset:column_offset + flatten_shape].to_numpy().reshape(
[len(data.index)] + shape), meta=param_meta)
column_offset += flatten_shape
return hapi_csv_file
return None