Module redvox.cloud.session_model_api

Session Models and API calls.

Sessions represent a summary of a single station from when it starts recording to when it stops recording.

Sessions are further subdivided into dynamic sessions. A single session contains keys to daily dynamic sessions. Daily dynamic sessions contain keys to hourly dynamic sessions. Hourly dynamic sessions contain keys to individual packets.

"""
Session Models and API calls.

Sessions represent a summary of a single station from when it starts recording to when it stops recording.

Sessions are further subdivided into dynamic sessions. A single session contains keys to daily dynamic sessions.
Daily dynamic sessions contain keys to hourly dynamic sessions.
Hourly dynamic sessions contain keys to individual packets.
"""

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List, Dict, Callable, Tuple, TYPE_CHECKING

import requests
from dataclasses_json import dataclass_json

from redvox.api1000.wrapped_redvox_packet.station_information import StationInformation
from redvox.api1000.wrapped_redvox_packet.timing_information import TimingInformation
from redvox.api1000.wrapped_redvox_packet.wrapped_packet import WrappedRedvoxPacketM
from redvox.cloud.api import post_req
from redvox.cloud.config import RedVoxConfig
from redvox.cloud.routes import RoutesV3
from redvox.common.date_time_utils import datetime_from_epoch_microseconds_utc as us2dt
from redvox.common.errors import RedVoxError

if TYPE_CHECKING:
    from redvox.cloud.client import CloudClient


@dataclass_json
@dataclass
class TimeSyncData:
    """
    Summarized time synchronization data.
    """

    ts: float
    lat: float
    off: float


@dataclass_json
@dataclass
class FirstLastBufTimeSync:
    """
    A bounded buffer that stores the first and last N samples of time synchronization data.
    """

    fst: List[Tuple[int, TimeSyncData]]
    fst_max_size: int
    lst: List[Tuple[int, TimeSyncData]]
    lst_max_size: int


@dataclass_json
@dataclass
class Timing:
    """
    High-level timing information.
    """

    first_data_ts: float
    last_data_ts: float
    n_ex: int
    mean_lat: float
    mean_off: float
    fst_lst: FirstLastBufTimeSync

    def first_data_dt(self) -> datetime:
        """
        :return: The datatime that represents the first datum.
        """
        return us2dt(self.first_data_ts)

    def last_data_dt(self) -> datetime:
        """
        :return: The datatime that represents the last datum.
        """
        return us2dt(self.last_data_ts)


@dataclass_json
@dataclass
class WelfordAggregator:
    """
    Fields required for computing the online Welford algorithm.
    See: https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm
    """

    m2: float
    mean: float
    cnt: int

    def finalize(self) -> Tuple[float, float, float]:
        """
        Computes the running variance.
        :return: A tuple containing the mean, variance, and sample variance.
        """
        if self.cnt < 2:
            raise RedVoxError("Not enough values to compute statistics from")
        variance: float = self.m2 / float(self.cnt)
        sample_variance: float = self.m2 / float(self.cnt - 1)
        return self.mean, variance, sample_variance


@dataclass_json
@dataclass
class Stats:
    """
    Contains the minimum and maximum values of a collection as well as the statistics provided by Welford's online
    algorithm.
    """

    min: float
    max: float
    welford: WelfordAggregator


@dataclass_json
@dataclass
class Sensor:
    """
    Describes a station's sensor.
    """

    name: str
    description: str
    sample_rate_stats: Stats


@dataclass_json
@dataclass
class Session:
    """
    Describes a single station's recording session.
    """

    id: str
    uuid: str
    desc: str
    start_ts: int
    client: str
    client_ver: str
    session_ver: str
    app: str
    api: int
    sub_api: int
    make: str
    model: str
    app_ver: str
    owner: str
    private: bool
    packet_dur: float
    sensors: List[Sensor]
    n_pkts: int
    timing: Timing
    sub: List[str]

    def session_key(self) -> str:
        """
        :return: The key associated with this session.
        """
        return f"{self.id}:{self.uuid}:{self.start_ts}"

    def query_dynamic_session(
        self, client: "CloudClient", sub: str
    ) -> "DynamicSessionModelResp":
        """
        Queries a dynamic session that is associated with this session.
        :param client: An instance of the cloud client.
        :param sub: The dynamic session key/
        :return: A DynamicSessionModelResp.
        """
        ts_parts: List[int] = list(map(int, sub.split(":")))
        return client.request_dynamic_session_model(
            self.session_key(), ts_parts[0], ts_parts[1]
        )


@dataclass_json
@dataclass
class Location:
    """
    Described a single location.
    """

    lat: float
    lng: float
    alt: float


@dataclass_json
@dataclass
class FirstLastBufLocation:
    """
    A bounded buffer that stores the first and last location points.
    """

    fst: List[Tuple[int, Location]]
    fst_max_size: int
    lst: List[Tuple[int, Location]]
    lst_max_size: int


@dataclass_json
@dataclass
class LocationStat:
    """
    Location statistics.
    """

    fst_lst: FirstLastBufLocation
    lat: Stats
    lng: Stats
    alt: Stats


@dataclass_json
@dataclass
class DynamicSession:
    """
    A dynamic session belongs to a parent session, but instead of representing an entire recording, dynamic sessions
    represent a chunk of time. Either daily or hourly dynamic sessions are supported.
    """

    n_pkts: int
    location: Dict[str, LocationStat]
    battery: Stats
    temperature: Stats

    session_key: str
    start_ts: int
    end_ts: int
    dur: str
    sub: List[str]

    def query_dynamic_session(
        self, client: "CloudClient", sub: str
    ) -> "DynamicSessionModelResp":
        """
        Queries a dynamic session that is associated with this session.
        :param client: An instance of the cloud client.
        :param sub: The dynamic session key.
        :return: A DynamicSessionModelResp.
        """
        ts_parts: List[int] = list(map(int, sub.split(":")))
        return client.request_dynamic_session_model(
            self.session_key, ts_parts[0], ts_parts[1]
        )

    # TODO
    def query_packet(self, client: "CloudClient", sub: str):
        """
        Queries individual packets assuming this is an hourly dynamic session.
        :param client: An instance of the cloud client.
        :param sub: The packet key.
        :return: TODO
        """
        raise RedVoxError("Method not yet implemented")


@dataclass_json
@dataclass
class SessionModelReq:
    """
    Request object for directly querying sessions.
    """

    auth_token: str
    session_key: str


@dataclass_json
@dataclass
class SessionModelResp:
    """
    Response of directly queried session.
    """

    err: Optional[str]
    session: Optional[Session]


@dataclass_json
@dataclass
class SessionModelsReq:
    """
    Request object for querying a range of sessions.
    """

    auth_token: str
    id_uuids: Optional[List[str]] = None
    owner: Optional[str] = None
    start_ts: Optional[int] = None
    end_ts: Optional[int] = None
    include_public: bool = False


@dataclass_json
@dataclass
class SessionModelsResp:
    """
    Response object from querying a range of sessions.
    """

    err: Optional[str]
    sessions: List[Session]


@dataclass_json
@dataclass
class DynamicSessionModelReq:
    """
    Request object for querying a dynamic session.
    """

    auth_token: str
    session_key: str
    start_ts: int
    end_ts: int


@dataclass_json
@dataclass
class DynamicSessionModelResp:
    """
    Response object of querying a dynamic session.
    """

    err: Optional[str]
    dynamic_session: Optional[DynamicSession]


def request_session(
    redvox_config: RedVoxConfig,
    req: SessionModelReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> SessionModelResp:
    """
    Requests a single session given the session key.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The session model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the SessionModelResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], SessionModelResp
    ] = lambda resp: SessionModelResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.SESSION_MODEL,
        req,
        handle_resp,
        session,
        timeout,
    )


def request_sessions(
    redvox_config: RedVoxConfig,
    req: SessionModelsReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> SessionModelsResp:
    """
    Requests a range of sessions.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The sessions model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the SessionModelsResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], SessionModelsResp
    ] = lambda resp: SessionModelsResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.SESSION_MODELS,
        req,
        handle_resp,
        session,
        timeout,
    )


def request_dynamic_session(
    redvox_config: RedVoxConfig,
    req: DynamicSessionModelReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> DynamicSessionModelResp:
    """
    Requests a single dynamic session given the dynamic session key.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The dynamic session model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the DynamicSessionModelResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], DynamicSessionModelResp
    ] = lambda resp: DynamicSessionModelResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.DYNAMIC_SESSION_MODEL,
        req,
        handle_resp,
        session,
        timeout,
    )


def session_key_from_packet(packet: WrappedRedvoxPacketM) -> str:
    """
    Constructs a session key given a RedVox packet.
    :param packet: The packet to construct a session key from.
    :return: A session key.
    """
    station_info: StationInformation = packet.get_station_information()
    if (
        station_info is None
        or station_info.get_id() == ""
        or station_info.get_uuid() == ""
    ):
        raise RedVoxError("Missing required station information")
    timing_info: TimingInformation = packet.get_timing_information()
    if timing_info is None or timing_info.get_app_start_mach_timestamp() == 0:
        raise RedVoxError("Missing required timing information")

    start_ts: int = round(timing_info.get_app_start_mach_timestamp())
    return f"{station_info.get_id()}:{station_info.get_uuid()}:{start_ts}"

Functions

def request_dynamic_session(redvox_config: RedVoxConfig, req: DynamicSessionModelReq, session: Optional[requests.sessions.Session] = None, timeout: Optional[float] = None) ‑> DynamicSessionModelResp

Requests a single dynamic session given the dynamic session key. :param redvox_config: An instance of the cloud configuration. :param req: The dynamic session model request. :param session: An optional HTTP client session. :param timeout: An optional timeout. :return: An instance of the DynamicSessionModelResp.

Expand source code

def request_dynamic_session(
    redvox_config: RedVoxConfig,
    req: DynamicSessionModelReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> DynamicSessionModelResp:
    """
    Requests a single dynamic session given the dynamic session key.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The dynamic session model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the DynamicSessionModelResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], DynamicSessionModelResp
    ] = lambda resp: DynamicSessionModelResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.DYNAMIC_SESSION_MODEL,
        req,
        handle_resp,
        session,
        timeout,
    )

def request_session(redvox_config: RedVoxConfig, req: SessionModelReq, session: Optional[requests.sessions.Session] = None, timeout: Optional[float] = None) ‑> SessionModelResp

Requests a single session given the session key. :param redvox_config: An instance of the cloud configuration. :param req: The session model request. :param session: An optional HTTP client session. :param timeout: An optional timeout. :return: An instance of the SessionModelResp.

Expand source code

def request_session(
    redvox_config: RedVoxConfig,
    req: SessionModelReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> SessionModelResp:
    """
    Requests a single session given the session key.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The session model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the SessionModelResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], SessionModelResp
    ] = lambda resp: SessionModelResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.SESSION_MODEL,
        req,
        handle_resp,
        session,
        timeout,
    )

def request_sessions(redvox_config: RedVoxConfig, req: SessionModelsReq, session: Optional[requests.sessions.Session] = None, timeout: Optional[float] = None) ‑> SessionModelsResp

Requests a range of sessions. :param redvox_config: An instance of the cloud configuration. :param req: The sessions model request. :param session: An optional HTTP client session. :param timeout: An optional timeout. :return: An instance of the SessionModelsResp.

Expand source code

def request_sessions(
    redvox_config: RedVoxConfig,
    req: SessionModelsReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> SessionModelsResp:
    """
    Requests a range of sessions.
    :param redvox_config: An instance of the cloud configuration.
    :param req: The sessions model request.
    :param session: An optional HTTP client session.
    :param timeout: An optional timeout.
    :return: An instance of the SessionModelsResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], SessionModelsResp
    ] = lambda resp: SessionModelsResp.from_dict(resp.json())
    return post_req(
        redvox_config,
        RoutesV3.SESSION_MODELS,
        req,
        handle_resp,
        session,
        timeout,
    )

def session_key_from_packet(packet: WrappedRedvoxPacketM) ‑> str

Constructs a session key given a RedVox packet. :param packet: The packet to construct a session key from. :return: A session key.

Expand source code

def session_key_from_packet(packet: WrappedRedvoxPacketM) -> str:
    """
    Constructs a session key given a RedVox packet.
    :param packet: The packet to construct a session key from.
    :return: A session key.
    """
    station_info: StationInformation = packet.get_station_information()
    if (
        station_info is None
        or station_info.get_id() == ""
        or station_info.get_uuid() == ""
    ):
        raise RedVoxError("Missing required station information")
    timing_info: TimingInformation = packet.get_timing_information()
    if timing_info is None or timing_info.get_app_start_mach_timestamp() == 0:
        raise RedVoxError("Missing required timing information")

    start_ts: int = round(timing_info.get_app_start_mach_timestamp())
    return f"{station_info.get_id()}:{station_info.get_uuid()}:{start_ts}"

Classes

class DynamicSession (n_pkts: int, location: Dict[str, LocationStat], battery: Stats, temperature: Stats, session_key: str, start_ts: int, end_ts: int, dur: str, sub: List[str])

A dynamic session belongs to a parent session, but instead of representing an entire recording, dynamic sessions represent a chunk of time. Either daily or hourly dynamic sessions are supported.

Expand source code

@dataclass_json
@dataclass
class DynamicSession:
    """
    A dynamic session belongs to a parent session, but instead of representing an entire recording, dynamic sessions
    represent a chunk of time. Either daily or hourly dynamic sessions are supported.
    """

    n_pkts: int
    location: Dict[str, LocationStat]
    battery: Stats
    temperature: Stats

    session_key: str
    start_ts: int
    end_ts: int
    dur: str
    sub: List[str]

    def query_dynamic_session(
        self, client: "CloudClient", sub: str
    ) -> "DynamicSessionModelResp":
        """
        Queries a dynamic session that is associated with this session.
        :param client: An instance of the cloud client.
        :param sub: The dynamic session key.
        :return: A DynamicSessionModelResp.
        """
        ts_parts: List[int] = list(map(int, sub.split(":")))
        return client.request_dynamic_session_model(
            self.session_key, ts_parts[0], ts_parts[1]
        )

    # TODO
    def query_packet(self, client: "CloudClient", sub: str):
        """
        Queries individual packets assuming this is an hourly dynamic session.
        :param client: An instance of the cloud client.
        :param sub: The packet key.
        :return: TODO
        """
        raise RedVoxError("Method not yet implemented")

Class variables

var battery : Stats

var dur : str

var end_ts : int

var location : Dict[str, LocationStat]

var n_pkts : int

var session_key : str

var start_ts : int

var sub : List[str]

var temperature : Stats

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def query_dynamic_session(self, client: CloudClient, sub: str) ‑> DynamicSessionModelResp

Queries a dynamic session that is associated with this session. :param client: An instance of the cloud client. :param sub: The dynamic session key. :return: A DynamicSessionModelResp.

Expand source code

def query_dynamic_session(
    self, client: "CloudClient", sub: str
) -> "DynamicSessionModelResp":
    """
    Queries a dynamic session that is associated with this session.
    :param client: An instance of the cloud client.
    :param sub: The dynamic session key.
    :return: A DynamicSessionModelResp.
    """
    ts_parts: List[int] = list(map(int, sub.split(":")))
    return client.request_dynamic_session_model(
        self.session_key, ts_parts[0], ts_parts[1]
    )

def query_packet(self, client: CloudClient, sub: str)

Queries individual packets assuming this is an hourly dynamic session. :param client: An instance of the cloud client. :param sub: The packet key. :return: TODO

Expand source code

def query_packet(self, client: "CloudClient", sub: str):
    """
    Queries individual packets assuming this is an hourly dynamic session.
    :param client: An instance of the cloud client.
    :param sub: The packet key.
    :return: TODO
    """
    raise RedVoxError("Method not yet implemented")

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class DynamicSessionModelReq (auth_token: str, session_key: str, start_ts: int, end_ts: int)

Request object for querying a dynamic session.

Expand source code

@dataclass_json
@dataclass
class DynamicSessionModelReq:
    """
    Request object for querying a dynamic session.
    """

    auth_token: str
    session_key: str
    start_ts: int
    end_ts: int

Class variables

var auth_token : str

var end_ts : int

var session_key : str

var start_ts : int

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class DynamicSessionModelResp (err: Optional[str], dynamic_session: Optional[DynamicSession])

Response object of querying a dynamic session.

Expand source code

@dataclass_json
@dataclass
class DynamicSessionModelResp:
    """
    Response object of querying a dynamic session.
    """

    err: Optional[str]
    dynamic_session: Optional[DynamicSession]

Class variables

var dynamic_session : Optional[DynamicSession]

var err : Optional[str]

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class FirstLastBufLocation (fst: List[Tuple[int, Location]], fst_max_size: int, lst: List[Tuple[int, Location]], lst_max_size: int)

A bounded buffer that stores the first and last location points.

Expand source code

@dataclass_json
@dataclass
class FirstLastBufLocation:
    """
    A bounded buffer that stores the first and last location points.
    """

    fst: List[Tuple[int, Location]]
    fst_max_size: int
    lst: List[Tuple[int, Location]]
    lst_max_size: int

Class variables

var fst : List[Tuple[int, Location]]

var fst_max_size : int

var lst : List[Tuple[int, Location]]

var lst_max_size : int

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class FirstLastBufTimeSync (fst: List[Tuple[int, TimeSyncData]], fst_max_size: int, lst: List[Tuple[int, TimeSyncData]], lst_max_size: int)

A bounded buffer that stores the first and last N samples of time synchronization data.

Expand source code

@dataclass_json
@dataclass
class FirstLastBufTimeSync:
    """
    A bounded buffer that stores the first and last N samples of time synchronization data.
    """

    fst: List[Tuple[int, TimeSyncData]]
    fst_max_size: int
    lst: List[Tuple[int, TimeSyncData]]
    lst_max_size: int

Class variables

var fst : List[Tuple[int, TimeSyncData]]

var fst_max_size : int

var lst : List[Tuple[int, TimeSyncData]]

var lst_max_size : int

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Location (lat: float, lng: float, alt: float)

Described a single location.

Expand source code

@dataclass_json
@dataclass
class Location:
    """
    Described a single location.
    """

    lat: float
    lng: float
    alt: float

Class variables

var alt : float

var lat : float

var lng : float

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class LocationStat (fst_lst: FirstLastBufLocation, lat: Stats, lng: Stats, alt: Stats)

Location statistics.

Expand source code

@dataclass_json
@dataclass
class LocationStat:
    """
    Location statistics.
    """

    fst_lst: FirstLastBufLocation
    lat: Stats
    lng: Stats
    alt: Stats

Class variables

var alt : Stats

var fst_lst : FirstLastBufLocation

var lat : Stats

var lng : Stats

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Sensor (name: str, description: str, sample_rate_stats: Stats)

Describes a station's sensor.

Expand source code

@dataclass_json
@dataclass
class Sensor:
    """
    Describes a station's sensor.
    """

    name: str
    description: str
    sample_rate_stats: Stats

Class variables

var description : str

var name : str

var sample_rate_stats : Stats

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Session (id: str, uuid: str, desc: str, start_ts: int, client: str, client_ver: str, session_ver: str, app: str, api: int, sub_api: int, make: str, model: str, app_ver: str, owner: str, private: bool, packet_dur: float, sensors: List[Sensor], n_pkts: int, timing: Timing, sub: List[str])

Describes a single station's recording session.

Expand source code

@dataclass_json
@dataclass
class Session:
    """
    Describes a single station's recording session.
    """

    id: str
    uuid: str
    desc: str
    start_ts: int
    client: str
    client_ver: str
    session_ver: str
    app: str
    api: int
    sub_api: int
    make: str
    model: str
    app_ver: str
    owner: str
    private: bool
    packet_dur: float
    sensors: List[Sensor]
    n_pkts: int
    timing: Timing
    sub: List[str]

    def session_key(self) -> str:
        """
        :return: The key associated with this session.
        """
        return f"{self.id}:{self.uuid}:{self.start_ts}"

    def query_dynamic_session(
        self, client: "CloudClient", sub: str
    ) -> "DynamicSessionModelResp":
        """
        Queries a dynamic session that is associated with this session.
        :param client: An instance of the cloud client.
        :param sub: The dynamic session key/
        :return: A DynamicSessionModelResp.
        """
        ts_parts: List[int] = list(map(int, sub.split(":")))
        return client.request_dynamic_session_model(
            self.session_key(), ts_parts[0], ts_parts[1]
        )

Class variables

var api : int

var app : str

var app_ver : str

var client : str

var client_ver : str

var desc : str

var id : str

var make : str

var model : str

var n_pkts : int

var owner : str

var packet_dur : float

var private : bool

var sensors : List[Sensor]

var session_ver : str

var start_ts : int

var sub : List[str]

var sub_api : int

var timing : Timing

var uuid : str

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def query_dynamic_session(self, client: CloudClient, sub: str) ‑> DynamicSessionModelResp

Queries a dynamic session that is associated with this session. :param client: An instance of the cloud client. :param sub: The dynamic session key/ :return: A DynamicSessionModelResp.

Expand source code

def query_dynamic_session(
    self, client: "CloudClient", sub: str
) -> "DynamicSessionModelResp":
    """
    Queries a dynamic session that is associated with this session.
    :param client: An instance of the cloud client.
    :param sub: The dynamic session key/
    :return: A DynamicSessionModelResp.
    """
    ts_parts: List[int] = list(map(int, sub.split(":")))
    return client.request_dynamic_session_model(
        self.session_key(), ts_parts[0], ts_parts[1]
    )

def session_key(self) ‑> str

:return: The key associated with this session.

Expand source code

def session_key(self) -> str:
    """
    :return: The key associated with this session.
    """
    return f"{self.id}:{self.uuid}:{self.start_ts}"

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class SessionModelReq (auth_token: str, session_key: str)

Request object for directly querying sessions.

Expand source code

@dataclass_json
@dataclass
class SessionModelReq:
    """
    Request object for directly querying sessions.
    """

    auth_token: str
    session_key: str

Class variables

var auth_token : str

var session_key : str

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class SessionModelResp (err: Optional[str], session: Optional[Session])

Response of directly queried session.

Expand source code

@dataclass_json
@dataclass
class SessionModelResp:
    """
    Response of directly queried session.
    """

    err: Optional[str]
    session: Optional[Session]

Class variables

var err : Optional[str]

var session : Optional[Session]

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class SessionModelsReq (auth_token: str, id_uuids: Optional[List[str]] = None, owner: Optional[str] = None, start_ts: Optional[int] = None, end_ts: Optional[int] = None, include_public: bool = False)

Request object for querying a range of sessions.

Expand source code

@dataclass_json
@dataclass
class SessionModelsReq:
    """
    Request object for querying a range of sessions.
    """

    auth_token: str
    id_uuids: Optional[List[str]] = None
    owner: Optional[str] = None
    start_ts: Optional[int] = None
    end_ts: Optional[int] = None
    include_public: bool = False

Class variables

var auth_token : str

var end_ts : Optional[int]

var id_uuids : Optional[List[str]]

var include_public : bool

var owner : Optional[str]

var start_ts : Optional[int]

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class SessionModelsResp (err: Optional[str], sessions: List[Session])

Response object from querying a range of sessions.

Expand source code

@dataclass_json
@dataclass
class SessionModelsResp:
    """
    Response object from querying a range of sessions.
    """

    err: Optional[str]
    sessions: List[Session]

Class variables

var err : Optional[str]

var sessions : List[Session]

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Stats (min: float, max: float, welford: WelfordAggregator)

Contains the minimum and maximum values of a collection as well as the statistics provided by Welford's online algorithm.

Expand source code

@dataclass_json
@dataclass
class Stats:
    """
    Contains the minimum and maximum values of a collection as well as the statistics provided by Welford's online
    algorithm.
    """

    min: float
    max: float
    welford: WelfordAggregator

Class variables

var max : float

var min : float

var welford : WelfordAggregator

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class TimeSyncData (ts: float, lat: float, off: float)

Summarized time synchronization data.

Expand source code

@dataclass_json
@dataclass
class TimeSyncData:
    """
    Summarized time synchronization data.
    """

    ts: float
    lat: float
    off: float

Class variables

var lat : float

var off : float

var ts : float

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Timing (first_data_ts: float, last_data_ts: float, n_ex: int, mean_lat: float, mean_off: float, fst_lst: FirstLastBufTimeSync)

High-level timing information.

Expand source code

@dataclass_json
@dataclass
class Timing:
    """
    High-level timing information.
    """

    first_data_ts: float
    last_data_ts: float
    n_ex: int
    mean_lat: float
    mean_off: float
    fst_lst: FirstLastBufTimeSync

    def first_data_dt(self) -> datetime:
        """
        :return: The datatime that represents the first datum.
        """
        return us2dt(self.first_data_ts)

    def last_data_dt(self) -> datetime:
        """
        :return: The datatime that represents the last datum.
        """
        return us2dt(self.last_data_ts)

Class variables

var first_data_ts : float

var fst_lst : FirstLastBufTimeSync

var last_data_ts : float

var mean_lat : float

var mean_off : float

var n_ex : int

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def first_data_dt(self) ‑> datetime.datetime

:return: The datatime that represents the first datum.

Expand source code

def first_data_dt(self) -> datetime:
    """
    :return: The datatime that represents the first datum.
    """
    return us2dt(self.first_data_ts)

def last_data_dt(self) ‑> datetime.datetime

:return: The datatime that represents the last datum.

Expand source code

def last_data_dt(self) -> datetime:
    """
    :return: The datatime that represents the last datum.
    """
    return us2dt(self.last_data_ts)

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class WelfordAggregator (m2: float, mean: float, cnt: int)

Fields required for computing the online Welford algorithm. See: https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm

Expand source code

@dataclass_json
@dataclass
class WelfordAggregator:
    """
    Fields required for computing the online Welford algorithm.
    See: https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm
    """

    m2: float
    mean: float
    cnt: int

    def finalize(self) -> Tuple[float, float, float]:
        """
        Computes the running variance.
        :return: A tuple containing the mean, variance, and sample variance.
        """
        if self.cnt < 2:
            raise RedVoxError("Not enough values to compute statistics from")
        variance: float = self.m2 / float(self.cnt)
        sample_variance: float = self.m2 / float(self.cnt - 1)
        return self.mean, variance, sample_variance

Class variables

var cnt : int

var m2 : float

var mean : float

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def finalize(self) ‑> Tuple[float, float, float]

Computes the running variance. :return: A tuple containing the mean, variance, and sample variance.

Expand source code

def finalize(self) -> Tuple[float, float, float]:
    """
    Computes the running variance.
    :return: A tuple containing the mean, variance, and sample variance.
    """
    if self.cnt < 2:
        raise RedVoxError("Not enough values to compute statistics from")
    variance: float = self.m2 / float(self.cnt)
    sample_variance: float = self.m2 / float(self.cnt - 1)
    return self.mean, variance, sample_variance

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)