Module redvox.cloud.station_stats

This module provides access to the timing station statistics through the RedVox Cloud API.

"""
This module provides access to the timing station statistics through the RedVox Cloud API.
"""

from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Callable, List, Optional, TypeVar

import requests
from dataclasses_json import dataclass_json

import redvox.common.file_statistics as file_stats
from redvox.cloud.api import post_req
from redvox.cloud.config import RedVoxConfig
from redvox.cloud.routes import RoutesV1
from redvox.common.date_time_utils import (
    datetime_from_epoch_microseconds_utc as us2dt,
    datetime_to_epoch_microseconds_utc as dt2us,
)

T = TypeVar("T")
R = TypeVar("R")


def _map_opt(opt: Optional[T], f: Callable[[T], R]) -> Optional[R]:
    """
    Maps an optional with the given function.
    :param opt: The optional top map.
    :param f: The mapping function.
    :return: The mapped value or None.
    """
    if opt is not None:
        return f(opt)
    return None


@dataclass_json
@dataclass
class GpsDateTime:
    """
    Represents timestamp pairings in API 1000 data.
    """

    mach_dt: float
    gps_dt: Optional[float]

    def into_file_stat(self) -> file_stats.GpsDateTime:
        """
        :return: Converts this into the file_statistics module representation.
        """
        gps_dt: Optional[datetime] = _map_opt(self.gps_dt, us2dt)
        return file_stats.GpsDateTime(us2dt(self.mach_dt), gps_dt)


@dataclass_json
@dataclass
class StationStat:
    """
    A collection of station fields for a given API 900 or API 1000 packet.
    These are used for timing correction and gap detection.
    """

    station_id: str
    station_uuid: str
    app_start_dt: Optional[float]
    packet_start_dt: float
    server_recv_dt: Optional[float]
    gps_dts: Optional[List[GpsDateTime]]
    latency: Optional[float]
    offset: Optional[float]
    sample_rate_hz: Optional[float]
    packet_duration: Optional[float]

    def into_file_stat(self) -> file_stats.StationStat:
        """
        :return: Converts this into the file_statistics module representation.
        """
        return file_stats.StationStat(
            self.station_id,
            self.station_uuid,
            _map_opt(self.app_start_dt, us2dt),
            us2dt(self.packet_start_dt),
            _map_opt(self.server_recv_dt, us2dt),
            _map_opt(
                self.gps_dts, lambda dts: list(map(GpsDateTime.into_file_stat, dts))
            ),
            self.latency,
            (self.packet_start_dt + (self.packet_duration / 2.0)),
            self.offset,
            self.sample_rate_hz,
            _map_opt(self.packet_duration, lambda dur: timedelta(microseconds=dur)),
        )


@dataclass_json
@dataclass
class StationStatReq:
    """
    A StationStatReq container.
    """

    auth_token: str
    start_ts_s: int
    end_ts_s: int
    station_ids: List[str]
    secret_token: Optional[str] = None


@dataclass
class StationStatsResp:
    """
    A response type converted into the file_statistics module representation.
    """

    station_stats: List[file_stats.StationStat]


@dataclass_json
@dataclass
class StationStatResp:
    """
    A response contain StationStat instances.
    """

    station_stats: List[StationStat]

    def into_station_stats_resp(self) -> StationStatsResp:
        return StationStatsResp(
            list(map(StationStat.into_file_stat, self.station_stats))
        )


def request_station_stats(
    redvox_config: RedVoxConfig,
    station_stat_req: StationStatReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> Optional[StationStatsResp]:
    """
    Requests timing statistics with the given parameters.
    :param redvox_config: The cloud configuration.
    :param station_stat_req: The request.
    :param session: The optional session.
    :param timeout: The optional timeout.
    :return: A StationStatsResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], StationStatResp
    ] = lambda resp: StationStatResp.from_dict(resp.json()).into_station_stats_resp()
    return post_req(
        redvox_config,
        RoutesV1.STATION_STATS,
        station_stat_req,
        handle_resp,
        session,
        timeout,
    )

Functions

def request_station_stats(redvox_config: RedVoxConfig, station_stat_req: StationStatReq, session: Optional[requests.sessions.Session] = None, timeout: Optional[float] = None) ‑> Optional[StationStatsResp]

Requests timing statistics with the given parameters. :param redvox_config: The cloud configuration. :param station_stat_req: The request. :param session: The optional session. :param timeout: The optional timeout. :return: A StationStatsResp.

Expand source code

def request_station_stats(
    redvox_config: RedVoxConfig,
    station_stat_req: StationStatReq,
    session: Optional[requests.Session] = None,
    timeout: Optional[float] = None,
) -> Optional[StationStatsResp]:
    """
    Requests timing statistics with the given parameters.
    :param redvox_config: The cloud configuration.
    :param station_stat_req: The request.
    :param session: The optional session.
    :param timeout: The optional timeout.
    :return: A StationStatsResp.
    """
    # noinspection Mypy
    handle_resp: Callable[
        [requests.Response], StationStatResp
    ] = lambda resp: StationStatResp.from_dict(resp.json()).into_station_stats_resp()
    return post_req(
        redvox_config,
        RoutesV1.STATION_STATS,
        station_stat_req,
        handle_resp,
        session,
        timeout,
    )

Classes

class GpsDateTime (mach_dt: float, gps_dt: Optional[float])

Represents timestamp pairings in API 1000 data.

Expand source code

@dataclass_json
@dataclass
class GpsDateTime:
    """
    Represents timestamp pairings in API 1000 data.
    """

    mach_dt: float
    gps_dt: Optional[float]

    def into_file_stat(self) -> file_stats.GpsDateTime:
        """
        :return: Converts this into the file_statistics module representation.
        """
        gps_dt: Optional[datetime] = _map_opt(self.gps_dt, us2dt)
        return file_stats.GpsDateTime(us2dt(self.mach_dt), gps_dt)

Class variables

var gps_dt : Optional[float]

var mach_dt : 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 into_file_stat(self) ‑> GpsDateTime

:return: Converts this into the file_statistics module representation.

Expand source code

def into_file_stat(self) -> file_stats.GpsDateTime:
    """
    :return: Converts this into the file_statistics module representation.
    """
    gps_dt: Optional[datetime] = _map_opt(self.gps_dt, us2dt)
    return file_stats.GpsDateTime(us2dt(self.mach_dt), gps_dt)

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 StationStat (station_id: str, station_uuid: str, app_start_dt: Optional[float], packet_start_dt: float, server_recv_dt: Optional[float], gps_dts: Optional[List[GpsDateTime]], latency: Optional[float], offset: Optional[float], sample_rate_hz: Optional[float], packet_duration: Optional[float])

A collection of station fields for a given API 900 or API 1000 packet. These are used for timing correction and gap detection.

Expand source code

@dataclass_json
@dataclass
class StationStat:
    """
    A collection of station fields for a given API 900 or API 1000 packet.
    These are used for timing correction and gap detection.
    """

    station_id: str
    station_uuid: str
    app_start_dt: Optional[float]
    packet_start_dt: float
    server_recv_dt: Optional[float]
    gps_dts: Optional[List[GpsDateTime]]
    latency: Optional[float]
    offset: Optional[float]
    sample_rate_hz: Optional[float]
    packet_duration: Optional[float]

    def into_file_stat(self) -> file_stats.StationStat:
        """
        :return: Converts this into the file_statistics module representation.
        """
        return file_stats.StationStat(
            self.station_id,
            self.station_uuid,
            _map_opt(self.app_start_dt, us2dt),
            us2dt(self.packet_start_dt),
            _map_opt(self.server_recv_dt, us2dt),
            _map_opt(
                self.gps_dts, lambda dts: list(map(GpsDateTime.into_file_stat, dts))
            ),
            self.latency,
            (self.packet_start_dt + (self.packet_duration / 2.0)),
            self.offset,
            self.sample_rate_hz,
            _map_opt(self.packet_duration, lambda dur: timedelta(microseconds=dur)),
        )

Class variables

var app_start_dt : Optional[float]

var gps_dts : Optional[List[GpsDateTime]]

var latency : Optional[float]

var offset : Optional[float]

var packet_duration : Optional[float]

var packet_start_dt : float

var sample_rate_hz : Optional[float]

var server_recv_dt : Optional[float]

var station_id : str

var station_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 into_file_stat(self) ‑> StationStat

:return: Converts this into the file_statistics module representation.

Expand source code

def into_file_stat(self) -> file_stats.StationStat:
    """
    :return: Converts this into the file_statistics module representation.
    """
    return file_stats.StationStat(
        self.station_id,
        self.station_uuid,
        _map_opt(self.app_start_dt, us2dt),
        us2dt(self.packet_start_dt),
        _map_opt(self.server_recv_dt, us2dt),
        _map_opt(
            self.gps_dts, lambda dts: list(map(GpsDateTime.into_file_stat, dts))
        ),
        self.latency,
        (self.packet_start_dt + (self.packet_duration / 2.0)),
        self.offset,
        self.sample_rate_hz,
        _map_opt(self.packet_duration, lambda dur: timedelta(microseconds=dur)),
    )

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 StationStatReq (auth_token: str, start_ts_s: int, end_ts_s: int, station_ids: List[str], secret_token: Optional[str] = None)

A StationStatReq container.

Expand source code

@dataclass_json
@dataclass
class StationStatReq:
    """
    A StationStatReq container.
    """

    auth_token: str
    start_ts_s: int
    end_ts_s: int
    station_ids: List[str]
    secret_token: Optional[str] = None

Class variables

var auth_token : str

var end_ts_s : int

var secret_token : Optional[str]

var start_ts_s : int

var station_ids : List[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 StationStatResp (station_stats: List[StationStat])

A response contain StationStat instances.

Expand source code

@dataclass_json
@dataclass
class StationStatResp:
    """
    A response contain StationStat instances.
    """

    station_stats: List[StationStat]

    def into_station_stats_resp(self) -> StationStatsResp:
        return StationStatsResp(
            list(map(StationStat.into_file_stat, self.station_stats))
        )

Class variables

var station_stats : List[StationStat]

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 into_station_stats_resp(self) ‑> StationStatsResp

Expand source code

def into_station_stats_resp(self) -> StationStatsResp:
    return StationStatsResp(
        list(map(StationStat.into_file_stat, self.station_stats))
    )

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 StationStatsResp (station_stats: List[StationStat])

A response type converted into the file_statistics module representation.

Expand source code

@dataclass
class StationStatsResp:
    """
    A response type converted into the file_statistics module representation.
    """

    station_stats: List[file_stats.StationStat]

Class variables

var station_stats : List[StationStat]