Skip to content

Time Series in FEWS PI format

See also: https://publicwiki.deltares.nl/display/FEWSDOC/Delft-Fews+Published+Interface+timeseries+Format+%28PI%29+Import

Events

Bases: DataFrame

FEWS-PI events in pandas DataFrame

Source code in src\fewspy\time_series.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class Events(pd.DataFrame):
    """FEWS-PI events in pandas DataFrame"""

    @classmethod
    def from_pi_events(
        cls, pi_events: list, missing_value: float = None, tz_offset: float = None
    ):
        """
        Parse Events from FEWS PI events dict.

        Args:
            pi_events (dict): FEWS PI events as dictionary

        Returns:
            Events: pandas DataFrame

        """

        df = cls(pi_events)

        # set datetime
        if tz_offset is not None:
            df["datetime"] = pd.to_datetime(
                df["date"] + " " + df["time"]
            ) - pd.Timedelta(hours=tz_offset)
        else:
            df["datetime"] = pd.to_datetime(df["date"] + " " + df["time"])

        # drop columns and add missing columns
        drop_cols = [i for i in df.columns if i not in EVENT_COLUMNS]
        df.drop(columns=drop_cols, inplace=True)
        for i in EVENT_COLUMNS:
            if i not in df.columns:
                df[i] = pd.NA

        # set numeric types
        df = df.astype(dtype={"value": "float32", "flag": "uint8"})

        # remove missings (if specified)
        if missing_value is not None:
            df = df.loc[df["value"] != missing_value]

        # set datetime to index
        df.set_index("datetime", inplace=True)

        return df

from_pi_events(pi_events, missing_value=None, tz_offset=None) classmethod

Parse Events from FEWS PI events dict.

Parameters:

Name Type Description Default
pi_events dict

FEWS PI events as dictionary

 required

Returns:

Name Type Description
Events

pandas DataFrame
Source code in src\fewspy\time_series.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@classmethod
def from_pi_events(
    cls, pi_events: list, missing_value: float = None, tz_offset: float = None
):
    """
    Parse Events from FEWS PI events dict.

    Args:
        pi_events (dict): FEWS PI events as dictionary

    Returns:
        Events: pandas DataFrame

    """

    df = cls(pi_events)

    # set datetime
    if tz_offset is not None:
        df["datetime"] = pd.to_datetime(
            df["date"] + " " + df["time"]
        ) - pd.Timedelta(hours=tz_offset)
    else:
        df["datetime"] = pd.to_datetime(df["date"] + " " + df["time"])

    # drop columns and add missing columns
    drop_cols = [i for i in df.columns if i not in EVENT_COLUMNS]
    df.drop(columns=drop_cols, inplace=True)
    for i in EVENT_COLUMNS:
        if i not in df.columns:
            df[i] = pd.NA

    # set numeric types
    df = df.astype(dtype={"value": "float32", "flag": "uint8"})

    # remove missings (if specified)
    if missing_value is not None:
        df = df.loc[df["value"] != missing_value]

    # set datetime to index
    df.set_index("datetime", inplace=True)

    return df

Header dataclass

FEWS-PI header-style dataclass

Source code in src\fewspy\time_series.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@dataclass
class Header:
    """FEWS-PI header-style dataclass"""

    type: str
    module_instance_id: str
    location_id: str
    parameter_id: str
    time_step: dict
    start_date: datetime
    end_date: datetime
    miss_val: float
    lat: float
    lon: float
    x: float
    y: float
    units: str
    station_name: str = None
    z: float = None
    qualifier_id: List[str] = None

    @classmethod
    def from_pi_header(cls, pi_header: dict):
        """
        Parse Header from FEWS PI header dict.

        Args:
            pi_header (dict): FEWS PI header as dictionary

        Returns:
            Header: FEWS-PI header-style dataclass

        """

        def _convert_kv(k: str, v) -> dict:
            k = camel_to_snake_case(k)
            if k in DATETIME_KEYS:
                v = dict_to_datetime(v)
            if k in FLOAT_KEYS:
                v = float(v)
            return k, v

        args = (_convert_kv(k, v) for k, v in pi_header.items())
        return cls(**{i[0]: i[1] for i in args})

from_pi_header(pi_header) classmethod

Parse Header from FEWS PI header dict.

Parameters:

Name Type Description Default
pi_header dict

FEWS PI header as dictionary

 required

Returns:

Name Type Description
Header

FEWS-PI header-style dataclass
Source code in src\fewspy\time_series.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def from_pi_header(cls, pi_header: dict):
    """
    Parse Header from FEWS PI header dict.

    Args:
        pi_header (dict): FEWS PI header as dictionary

    Returns:
        Header: FEWS-PI header-style dataclass

    """

    def _convert_kv(k: str, v) -> dict:
        k = camel_to_snake_case(k)
        if k in DATETIME_KEYS:
            v = dict_to_datetime(v)
        if k in FLOAT_KEYS:
            v = float(v)
        return k, v

    args = (_convert_kv(k, v) for k, v in pi_header.items())
    return cls(**{i[0]: i[1] for i in args})

TimeSeries dataclass

FEWS-PI time series

Source code in src\fewspy\time_series.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@dataclass
class TimeSeries:
    """FEWS-PI time series"""

    header: Header
    events: Events = pd.DataFrame(columns=EVENT_COLUMNS).set_index("datetime")

    @classmethod
    def from_pi_time_series(cls, pi_time_series: dict, time_zone: float = None):
        """Parse TimeSeries from FEWS PI timeseries dict.

        Args:
            pi_time_series (dict): FEWS PI timeseries as dictionary
            time_zone (float, optional): time_zone. Defaults to None.

        Returns:
            fewspy.TimeSeries: time series in FEWS PI format
        """
        header = Header.from_pi_header(pi_time_series["header"])
        kwargs = dict(header=header)
        if "events" in pi_time_series.keys():
            kwargs["events"] = Events.from_pi_events(
                pi_time_series["events"], header.miss_val, time_zone
            )
        return cls(**kwargs)

from_pi_time_series(pi_time_series, time_zone=None) classmethod

Parse TimeSeries from FEWS PI timeseries dict.

Parameters:

Name Type Description Default
pi_time_series dict

FEWS PI timeseries as dictionary

 required
time_zone float

time_zone. Defaults to None.

 None

Returns:

Type Description

fewspy.TimeSeries: time series in FEWS PI format
Source code in src\fewspy\time_series.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def from_pi_time_series(cls, pi_time_series: dict, time_zone: float = None):
    """Parse TimeSeries from FEWS PI timeseries dict.

    Args:
        pi_time_series (dict): FEWS PI timeseries as dictionary
        time_zone (float, optional): time_zone. Defaults to None.

    Returns:
        fewspy.TimeSeries: time series in FEWS PI format
    """
    header = Header.from_pi_header(pi_time_series["header"])
    kwargs = dict(header=header)
    if "events" in pi_time_series.keys():
        kwargs["events"] = Events.from_pi_events(
            pi_time_series["events"], header.miss_val, time_zone
        )
    return cls(**kwargs)

TimeSeriesSet dataclass

FEWS-PI time series set

Source code in src\fewspy\time_series.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
@dataclass
class TimeSeriesSet:
    """FEWS-PI time series set"""
    version: str = None
    time_zone: float = None
    time_series: List[TimeSeries] = field(default_factory=list)

    def __len__(self):
        return len(self.time_series)

    @classmethod
    def from_pi_time_series(cls, pi_time_series_set: dict):
        """Parse TimeSeries from FEWS PI time series set dict.

        Args:
            pi_time_series_set (dict): FEWS PI time series set as dictionary

        Returns:
            fewspy.TimeSeriesSet: Time series set with multiple time series
        """
        kwargs = {}
        if "version" in pi_time_series_set.keys():
            kwargs["version"] = pi_time_series_set["version"]
        if "timeZone" in pi_time_series_set.keys():
            time_zone = float(pi_time_series_set["timeZone"])
            kwargs["time_zone"] = time_zone
        if "timeSeries" in pi_time_series_set.keys():
            kwargs["time_series"] = [
                TimeSeries.from_pi_time_series(i, time_zone)
                for i in pi_time_series_set["timeSeries"]
            ]
        return cls(**kwargs)

    def add(self, time_series_set):
        # add time_series_set to the time_series_set
        return self

    @property
    def empty(self):
        return all([i.events.empty for i in self.time_series])

    @property
    def parameter_ids(self):
        return list(set([i.header.parameter_id for i in self.time_series]))

    @property
    def location_ids(self):
        return list(set([i.header.location_id for i in self.time_series]))

    @property
    def qualifier_ids(self):
        qualifiers = (i.header.qualifier_id for i in self.time_series)
        qualifiers = [i for i in qualifiers if i is not None]

        return list(set(flatten_list(qualifiers)))

from_pi_time_series(pi_time_series_set) classmethod

Parse TimeSeries from FEWS PI time series set dict.

Parameters:

Name Type Description Default
pi_time_series_set dict

FEWS PI time series set as dictionary

 required

Returns:

Type Description

fewspy.TimeSeriesSet: Time series set with multiple time series
Source code in src\fewspy\time_series.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def from_pi_time_series(cls, pi_time_series_set: dict):
    """Parse TimeSeries from FEWS PI time series set dict.

    Args:
        pi_time_series_set (dict): FEWS PI time series set as dictionary

    Returns:
        fewspy.TimeSeriesSet: Time series set with multiple time series
    """
    kwargs = {}
    if "version" in pi_time_series_set.keys():
        kwargs["version"] = pi_time_series_set["version"]
    if "timeZone" in pi_time_series_set.keys():
        time_zone = float(pi_time_series_set["timeZone"])
        kwargs["time_zone"] = time_zone
    if "timeSeries" in pi_time_series_set.keys():
        kwargs["time_series"] = [
            TimeSeries.from_pi_time_series(i, time_zone)
            for i in pi_time_series_set["timeSeries"]
        ]
    return cls(**kwargs)

reliables(df, threshold=6)

Filters reliables from an Events type Pandas DataFrame

Parameters:

Name Type Description Default
df DataFrame

input Events-type Pandas Dataframe

 required
threshold int

threshold for unreleables. Defaults to 6.

 6

Returns:

Type Description
DataFrame

pd.DataFrame: Pandas DataFrame with reliable data only
Source code in src\fewspy\time_series.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def reliables(df: pd.DataFrame, threshold: int = 6) -> pd.DataFrame:
    """
    Filters reliables from an Events type Pandas DataFrame

    Args:
        df (pd.DataFrame): input Events-type Pandas Dataframe
        threshold (int, optional): threshold for unreleables. Defaults to 6.

    Returns:
        pd.DataFrame: Pandas DataFrame with reliable data only

    """

    return df.loc[df["flag"] < threshold]