Skip to content

dp3.common.datapoint

DataPointBase

Bases: BaseModel

Data-point

Contains single raw data value received on API. This is just base class - plain, observation or timeseries datapoints inherit from this class (see below).

Provides front line of validation for this data value.

Internal usage: inside Task, created by TaskExecutor

DataPointPlainBase

Bases: DataPointBase

Plain attribute data-point

Contains single raw data value received on API for plain attribute.

In case of plain data-point, it's not really a data-point, but we use the same naming for simplicity.

DataPointObservationsBase

Bases: DataPointBase

Observations attribute data-point

Contains single raw data value received on API for observations attribute.

DataPointTimeseriesBase

Bases: DataPointBase

Timeseries attribute data-point

Contains single raw data value received on API for observations attribute.

ignore_value 

ignore_value(_v)

Ignore the passed value and return None.

Source code in dp3/common/datapoint.py 
17
18
19
def ignore_value(_v):
    """Ignore the passed value and return None."""
    return None

is_list_ordered 

is_list_ordered(to_check: list)

Checks if list is ordered (not decreasing anywhere)

Source code in dp3/common/datapoint.py 
78
79
80
def is_list_ordered(to_check: list):
    """Checks if list is ordered (not decreasing anywhere)"""
    return all(to_check[i] <= to_check[i + 1] for i in range(len(to_check) - 1))

dp_ts_v_validator 

dp_ts_v_validator(v)

Check if all value arrays are the same length.

Source code in dp3/common/datapoint.py 
87
88
89
90
91
92
def dp_ts_v_validator(v):
    """Check if all value arrays are the same length."""
    values_len = {len(v_i) for _, v_i in v.model_dump().items()}
    assert len(values_len) == 1, f"Series values have different lengths: {values_len}"

    return v

dp_ts_root_validator_irregular 

dp_ts_root_validator_irregular(self)

Validates or sets t2 of irregular timeseries datapoint

Source code in dp3/common/datapoint.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def dp_ts_root_validator_irregular(self):
    """Validates or sets t2 of irregular timeseries datapoint"""
    first_time = self.v.time[0]
    last_time = self.v.time[-1]

    # Check t1 <= first_time
    assert self.t1 <= first_time, f"'t1' is above first item in 'time' series ({first_time})"

    # Check last_time <= t2
    if self.t2:
        assert self.t2 >= last_time, f"'t2' is below last item in 'time' series ({last_time})"
    else:
        self.t2 = last_time

    # time must be ordered
    assert is_list_ordered(self.v.time), "'time' series is not ordered"

    return self

dp_ts_root_validator_irregular_intervals 

dp_ts_root_validator_irregular_intervals(self)

Validates or sets t2 of irregular intervals timeseries datapoint

Source code in dp3/common/datapoint.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def dp_ts_root_validator_irregular_intervals(self):
    """Validates or sets t2 of irregular intervals timeseries datapoint"""
    first_time = self.v.time_first[0]
    last_time = self.v.time_last[-1]

    # Check t1 <= first_time
    assert self.t1 <= first_time, f"'t1' is above first item in 'time_first' series ({first_time})"

    # Check last_time <= t2
    if self.t2:
        assert self.t2 >= last_time, f"'t2' is below last item in 'time_last' series ({last_time})"
    else:
        self.t2 = last_time

    # Check time_first[i] <= time_last[i]
    assert all(
        t[0] <= t[1] for t in zip(self.v.time_first, self.v.time_last)
    ), "'time_first[i] <= time_last[i]' isn't true for all 'i'"

    return self