Module redvox.common.station_model

Prototype functions for station location and timing summary Provides an example of how to access the information from a Station object This module will contain methods to summarize a single Station

"""
Prototype functions for station location and timing summary
Provides an example of how to access the information from a Station object
This module will contain methods to summarize a single Station
"""
from typing import List, Optional
import math

from dataclasses import dataclass
from dataclasses_json import dataclass_json
import numpy as np

from redvox.common.station import Station
from redvox.api1000.wrapped_redvox_packet.sensors.location import LocationProvider


LOCATION_PROVIDER_IDX: int = 10
LATITUDE_IDX: int = 1
LONGITUDE_IDX: int = 2
ALTITUDE_IDX: int = 3


@dataclass_json
@dataclass
class LocationSummary:
    """
    Summary of a Location.  Includes the location sensor's name, source of data (provider), number of points,
    and the mean and standard deviation of latitude, longitude, and altitude.
    """

    name: str
    provider: str
    latitude_mean: float
    latitude_standard_deviation: float
    longitude_mean: float
    longitude_standard_deviation: float
    altitude_mean: Optional[float]
    altitude_standard_deviation: float
    num_pts: int

    @staticmethod
    def from_station(stn: Station) -> Optional[List["LocationSummary"]]:
        """
        :param stn: Station to get data from
        :return: List of LocationSummary grouped by provider or None
        """
        loc = stn.find_loc_for_stats()
        if loc:
            samples: np.ndarray = loc.samples()

            if loc.num_samples() == 1:
                return [
                    LocationSummary(
                        loc.name,
                        LocationProvider(int(samples[LOCATION_PROVIDER_IDX][0])).name,
                        samples[LATITUDE_IDX][0],
                        0.0,
                        samples[LONGITUDE_IDX][0],
                        0.0,
                        samples[ALTITUDE_IDX][0],
                        0.0,
                        1,
                    )
                ]

            # for each provider, create a new entry in the dictionary or append new data
            loc_prov_to_data: dict = {}
            lat_samples: np.ndarray = samples[LATITUDE_IDX]
            lng_samples: np.ndarray = samples[LONGITUDE_IDX]
            for j in range(min(len(lat_samples), len(lng_samples))):
                if (not math.isnan(lat_samples[j])) and (not math.isnan(lng_samples[j])):
                    prov = LocationProvider(int(samples[LOCATION_PROVIDER_IDX][j])).name
                    data = np.array(
                        [
                            [samples[LATITUDE_IDX][j]],
                            [samples[LONGITUDE_IDX][j]],
                            [samples[ALTITUDE_IDX][j]],
                        ]
                    )
                    loc_prov_to_data[prov] = (
                        data
                        if prov not in loc_prov_to_data.keys()
                        else np.concatenate([loc_prov_to_data[prov], data], 1)
                    )

            loc_sums = []
            for provider, loc_data in loc_prov_to_data.items():
                loc_sums.append(
                    LocationSummary(
                        loc.name,
                        provider,
                        loc_data[0].mean(),
                        loc_data[0].std(),
                        loc_data[1].mean(),
                        loc_data[1].std(),
                        loc_data[2].mean(),
                        loc_data[2].std(),
                        len(loc_data[0]),
                    )
                )
            return loc_sums
        return None


def example_of_print_loc_and_timing_summary(stn: Station):
    """
    An example function that show how you can access location, GNSS, and time sync information and prints it
    :param stn: Station to get data from
    """
    print(f"station_id: {stn.id()}")
    loc_summary = LocationSummary.from_station(stn)
    if loc_summary:
        for m in loc_summary:
            print(
                "-----------------------------\n"
                f"location_provider: {m.provider}, num_pts: {m.num_pts}\n"
                f"latitude  mean: {m.latitude_mean}, std_dev: {m.latitude_standard_deviation}\n"
                f"longitude mean: {m.longitude_mean}, std_dev: {m.longitude_standard_deviation}\n"
                f"altitude  mean: {m.altitude_mean}, std_dev: {m.altitude_standard_deviation}"
            )
        print(
            f"-----------------------------\nGNSS timing:\n"
            f"GNSS offset at start: {stn.gps_offset_model().intercept}\n"
            f"slope: {stn.gps_offset_model().slope}\n"
            f"GNSS estimated latency: {stn.gps_offset_model().mean_latency}\n"
            f"num_pts: {len(stn.location_sensor().get_gps_timestamps_data())}"
        )
    else:
        print("location data not found.")
    print(
        f"-----------------------------\ntimesync:\n"
        f"best_offset: {stn.timesync_data().best_offset()}, offset_std: {stn.timesync_data().offset_std()}\n"
        f"best_latency: {stn.timesync_data().best_latency()}, latency_std: {stn.timesync_data().latency_std()}\n"
        f"num_pts: {stn.timesync_data().num_tri_messages()}\n*****************************\n"
    )

Functions

def example_of_print_loc_and_timing_summary(stn: Station)

An example function that show how you can access location, GNSS, and time sync information and prints it :param stn: Station to get data from

Expand source code

def example_of_print_loc_and_timing_summary(stn: Station):
    """
    An example function that show how you can access location, GNSS, and time sync information and prints it
    :param stn: Station to get data from
    """
    print(f"station_id: {stn.id()}")
    loc_summary = LocationSummary.from_station(stn)
    if loc_summary:
        for m in loc_summary:
            print(
                "-----------------------------\n"
                f"location_provider: {m.provider}, num_pts: {m.num_pts}\n"
                f"latitude  mean: {m.latitude_mean}, std_dev: {m.latitude_standard_deviation}\n"
                f"longitude mean: {m.longitude_mean}, std_dev: {m.longitude_standard_deviation}\n"
                f"altitude  mean: {m.altitude_mean}, std_dev: {m.altitude_standard_deviation}"
            )
        print(
            f"-----------------------------\nGNSS timing:\n"
            f"GNSS offset at start: {stn.gps_offset_model().intercept}\n"
            f"slope: {stn.gps_offset_model().slope}\n"
            f"GNSS estimated latency: {stn.gps_offset_model().mean_latency}\n"
            f"num_pts: {len(stn.location_sensor().get_gps_timestamps_data())}"
        )
    else:
        print("location data not found.")
    print(
        f"-----------------------------\ntimesync:\n"
        f"best_offset: {stn.timesync_data().best_offset()}, offset_std: {stn.timesync_data().offset_std()}\n"
        f"best_latency: {stn.timesync_data().best_latency()}, latency_std: {stn.timesync_data().latency_std()}\n"
        f"num_pts: {stn.timesync_data().num_tri_messages()}\n*****************************\n"
    )

Classes

class LocationSummary (name: str, provider: str, latitude_mean: float, latitude_standard_deviation: float, longitude_mean: float, longitude_standard_deviation: float, altitude_mean: Optional[float], altitude_standard_deviation: float, num_pts: int)

Summary of a Location. Includes the location sensor's name, source of data (provider), number of points, and the mean and standard deviation of latitude, longitude, and altitude.

Expand source code

@dataclass_json
@dataclass
class LocationSummary:
    """
    Summary of a Location.  Includes the location sensor's name, source of data (provider), number of points,
    and the mean and standard deviation of latitude, longitude, and altitude.
    """

    name: str
    provider: str
    latitude_mean: float
    latitude_standard_deviation: float
    longitude_mean: float
    longitude_standard_deviation: float
    altitude_mean: Optional[float]
    altitude_standard_deviation: float
    num_pts: int

    @staticmethod
    def from_station(stn: Station) -> Optional[List["LocationSummary"]]:
        """
        :param stn: Station to get data from
        :return: List of LocationSummary grouped by provider or None
        """
        loc = stn.find_loc_for_stats()
        if loc:
            samples: np.ndarray = loc.samples()

            if loc.num_samples() == 1:
                return [
                    LocationSummary(
                        loc.name,
                        LocationProvider(int(samples[LOCATION_PROVIDER_IDX][0])).name,
                        samples[LATITUDE_IDX][0],
                        0.0,
                        samples[LONGITUDE_IDX][0],
                        0.0,
                        samples[ALTITUDE_IDX][0],
                        0.0,
                        1,
                    )
                ]

            # for each provider, create a new entry in the dictionary or append new data
            loc_prov_to_data: dict = {}
            lat_samples: np.ndarray = samples[LATITUDE_IDX]
            lng_samples: np.ndarray = samples[LONGITUDE_IDX]
            for j in range(min(len(lat_samples), len(lng_samples))):
                if (not math.isnan(lat_samples[j])) and (not math.isnan(lng_samples[j])):
                    prov = LocationProvider(int(samples[LOCATION_PROVIDER_IDX][j])).name
                    data = np.array(
                        [
                            [samples[LATITUDE_IDX][j]],
                            [samples[LONGITUDE_IDX][j]],
                            [samples[ALTITUDE_IDX][j]],
                        ]
                    )
                    loc_prov_to_data[prov] = (
                        data
                        if prov not in loc_prov_to_data.keys()
                        else np.concatenate([loc_prov_to_data[prov], data], 1)
                    )

            loc_sums = []
            for provider, loc_data in loc_prov_to_data.items():
                loc_sums.append(
                    LocationSummary(
                        loc.name,
                        provider,
                        loc_data[0].mean(),
                        loc_data[0].std(),
                        loc_data[1].mean(),
                        loc_data[1].std(),
                        loc_data[2].mean(),
                        loc_data[2].std(),
                        len(loc_data[0]),
                    )
                )
            return loc_sums
        return None

Class variables

var altitude_mean : Optional[float]

var altitude_standard_deviation : float

var latitude_mean : float

var latitude_standard_deviation : float

var longitude_mean : float

var longitude_standard_deviation : float

var name : str

var num_pts : int

var provider : 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 from_station(stn: Station) ‑> Optional[List[LocationSummary]]

:param stn: Station to get data from :return: List of LocationSummary grouped by provider or None

Expand source code

@staticmethod
def from_station(stn: Station) -> Optional[List["LocationSummary"]]:
    """
    :param stn: Station to get data from
    :return: List of LocationSummary grouped by provider or None
    """
    loc = stn.find_loc_for_stats()
    if loc:
        samples: np.ndarray = loc.samples()

        if loc.num_samples() == 1:
            return [
                LocationSummary(
                    loc.name,
                    LocationProvider(int(samples[LOCATION_PROVIDER_IDX][0])).name,
                    samples[LATITUDE_IDX][0],
                    0.0,
                    samples[LONGITUDE_IDX][0],
                    0.0,
                    samples[ALTITUDE_IDX][0],
                    0.0,
                    1,
                )
            ]

        # for each provider, create a new entry in the dictionary or append new data
        loc_prov_to_data: dict = {}
        lat_samples: np.ndarray = samples[LATITUDE_IDX]
        lng_samples: np.ndarray = samples[LONGITUDE_IDX]
        for j in range(min(len(lat_samples), len(lng_samples))):
            if (not math.isnan(lat_samples[j])) and (not math.isnan(lng_samples[j])):
                prov = LocationProvider(int(samples[LOCATION_PROVIDER_IDX][j])).name
                data = np.array(
                    [
                        [samples[LATITUDE_IDX][j]],
                        [samples[LONGITUDE_IDX][j]],
                        [samples[ALTITUDE_IDX][j]],
                    ]
                )
                loc_prov_to_data[prov] = (
                    data
                    if prov not in loc_prov_to_data.keys()
                    else np.concatenate([loc_prov_to_data[prov], data], 1)
                )

        loc_sums = []
        for provider, loc_data in loc_prov_to_data.items():
            loc_sums.append(
                LocationSummary(
                    loc.name,
                    provider,
                    loc_data[0].mean(),
                    loc_data[0].std(),
                    loc_data[1].mean(),
                    loc_data[1].std(),
                    loc_data[2].mean(),
                    loc_data[2].std(),
                    len(loc_data[0]),
                )
            )
        return loc_sums
    return None

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)