Reference

jointly.synchronizer

class jointly.synchronizer.Synchronizer(sources: Dict[str, Dict[str, Optional[Union[str, pandas.core.frame.DataFrame, float, pandas._libs.tslibs.timedeltas.Timedelta]]]], reference_source_name: str, extractor: Optional[jointly.abstract_extractor.AbstractExtractor] = None, sampling_freq: Optional[float] = None)[source]

Bases: object

property extractor: jointly.abstract_extractor.AbstractExtractor

Get the current extractor

get_sync_params(recalculate: bool = False)[source]

Get the synchronization params. If they have not been calculated yet, they will be.

Parameters

recalculate – force calculation, even if it was already done before

Returns

the synchronization params for each source, i.e., each timeshift and stretch factor

get_synced_data(recalculate: bool = False) Dict[str, pandas.core.frame.DataFrame][source]

Synchronize the input data.

Parameters

recalculate – force recalculating the synchronization parameters

Returns

a dictionary of the shifted and stretched source signals

save_data(target_dir: str, tables: Optional[Dict[str, Dict[str, List[str]]]] = None, save_total_table: bool = True)[source]

Export synchronized data. Two formats are possible: if tables is given, a file for each root key is created containing the columns from the sensors specified as the keys on the second level. This can be used to create a file for each sensor type, see ResultTableSpec for an example.

A SYNC.csv is always exported to store the synchronization parameters that have been calculated.

Parameters
  • target_dir – target directory for the export files

  • tables – ResultTableSpec to specify the export format, or None

  • save_total_table – exports an outer join over all synchronized dataframes

save_pickles(target_dir: str) Dict[str, pandas.core.frame.DataFrame][source]

Save a pickled, synced, dataframe for each source file. Does not save a total table. Sync parameters are saved as SYNC.csv.

Parameters

target_dir – target directory for the export files

Returns

the synced data, plus a sync parameter dataframe in the dictionary entry with the key “SYNC”.

jointly.shake_extractor

class jointly.shake_extractor.ShakeExtractor[source]

Bases: jointly.abstract_extractor.AbstractExtractor

distance = 1500

distance in milliseconds in which the next peak must occur to be considered a sequence

property end_window_length: pandas._libs.tslibs.timedeltas.Timedelta

time window as pandas.Timedelta in which to look for peaks at end of signal

get_segments(signals: pandas.core.frame.DataFrame) Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]][source]

Returns dictionary with start and end for each sensor source, i.e., a SyncPairs instance

Parameters

signals – DataFrame containing the reference signals for each source

Returns

SyncPairs instance

min_length = 6

minimum number of peaks per sequence

property start_window_length: pandas._libs.tslibs.timedeltas.Timedelta

time window as pandas.Timedelta in which to look for peaks from start of signal

property threshold: float

min height for peak detection. In range [0, 1], as the data is normalized

time_buffer = Timedelta('0 days 00:00:01')

time in seconds will be padded to first and last peak for timestamps of segment

jointly.types

jointly.types.ResultTableSpec

Specification for saving the synchronized results in separated files, with each root key defining a target file. The second level defines the columns which should be saved from each source file into the given target file. This can be used to separate the input files into files containing only a single sensor type, e.g., to extract the PPG signal from two different sensors into a single file.

Example:

{
    'ACC': {
        'Faros': ['Accelerometer_X', 'Accelerometer_Y', 'Accelerometer_Z'],
        'Empatica': ['acc_x', 'acc_y', 'acc_z'],
        'Everion': ['accx_data', 'accy_data', 'accz_data'],
    },
    'PPG': {
        'Empatica': ['bvp'],
        'Everion': ['blood_pulse_wave', 'led2_data', 'led3_data'],
    },
    'EDA': {
        'Empatica': ['eda'],
        'Everion': ['gsr_electrode'],
    },
    'ECG': {
        'Faros': ['ECG'],
    },
    'TEMP': {
        'Empatica': ['temp'],
        'Everion': ['temperature_object'],
    },
    'HR': {
        'Empatica': ['hr'],
        'Everion': ['heart_rate', 'heart_rate_quality'],
    },
    'IBI': {
        'Faros': ['HRV'],
        'Empatica': ['ibi'],
        'Everion': ['inter_pulse_interval', 'inter_pulse_interval_deviation'],
    }
}

alias of Dict[str, Dict[str, List[str]]]

jointly.types.SourceDict

A dictionary of dictionaries. Each entry defines an input sensor, and points to a dictionary with the keys data and ref_column.

data is a pandas DataFrame with a DateTimeIndex.

ref_column specifies the column within data which should be used to extract synchronization points, e.g., shakes.

alias of Dict[str, Dict[str, Optional[Union[str, pandas.core.frame.DataFrame, float, pandas._libs.tslibs.timedeltas.Timedelta]]]]

jointly.types.SyncPairTimeshift

Timeshift for a single sync pair, i.e., the shift required to synchronize one pair to the reference signal

alias of Dict[str, pandas._libs.tslibs.timedeltas.Timedelta]

jointly.types.SyncPairs

A dictionary that contains SynchronizationPair instances for a number of sources.

alias of Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]]

jointly.types.SynchronizationPair

A dictionary containing both the first and the second synchronization point of a signal. Two points are required to calculate the distance in between them. Properties are first and second.

alias of Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]

jointly.types.SynchronizationPoint

A dictionary describing a synchronization point, e.g., a shake. A synchronization point has a start and an end, and thus the properties start and end.

alias of Dict[str, pandas._libs.tslibs.timestamps.Timestamp]

jointly.synchronization_errors

exception jointly.synchronization_errors.BadThresholdException[source]

Bases: Exception

Thrown if the shake threshold is below 0 or above 1.

exception jointly.synchronization_errors.BadWindowException[source]

Bases: Exception

Thrown when the sync point detection window length is longer than the data

exception jointly.synchronization_errors.ShakeMissingException[source]

Bases: Exception

Thrown when a synchronization point is missing, e.g., a second shake could not be found in the signal.

exception jointly.synchronization_errors.StartEqualsEndError[source]

Bases: Exception

Thrown when the detected start synchronization point equals the end synchronization point. Maybe change the detection window lengths?

jointly.helpers

Contains various helper functions useful in conjunction with or internally to jointly.

jointly.helpers.calculate_magnitude(df: pandas.core.frame.DataFrame, of_cols: List[str], title: str = 'Magnitude') pandas.core.frame.DataFrame[source]

Calculate the magnitude of a subset of columns from a DataFrame.

Will return 0 if the input data is NaN to allow future algorithms to continue working despite the NaN in the input values.

jointly.helpers.get_equidistant_signals(signals: pandas.core.frame.DataFrame, frequency: float)[source]

Returns dataframe with columns from signals sampled equidistantly at the specified frequency.

Parameters
  • signals – the columns of this dataframe will be independently resampled

  • frequency – the target frequency in Hz

Returns

equidistantly sampled dataframe

jointly.helpers.get_max_ref_frequency(signals: pandas.core.frame.DataFrame) float[source]

Get the maximum frequency in the dataframe

Parameters

signals – input dataframe with the given signals

Returns

float describing the maximum frequency in the source data.

jointly.helpers.get_segment_data(dataframe: pandas.core.frame.DataFrame, segments: Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]], col: str, segment: str) Tuple[pandas._libs.tslibs.timestamps.Timestamp, pandas._libs.tslibs.timestamps.Timestamp, pandas.core.frame.DataFrame][source]

Return a 3-tuple of start and end indices plus data within that timeframe of the given column in the given dataframe.

jointly.helpers.get_stretch_factor(segments: Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]], timeshifts: Dict[str, pandas._libs.tslibs.timedeltas.Timedelta]) float[source]

Get the stretch factor required to stretch the duration between segments such that it will fit exactly to the signal when shifted by the amount given by timeshifts.

This is a function that should basically exclusively be used within jointly, as the parameters are produced during the synchronization process.

Parameters
  • segments – the segment instance containing the segment info to be stretched

  • timeshifts – the timeshifts that should be applied to make the signal align to the reference signal

Returns

a float as described above

jointly.helpers.infer_freq(series: pandas.core.series.Series) float[source]

Infer the frequency of a series by finding the median temporal distance between its elements.

Parameters

series – the frequency of this series will be inferred

Returns

frequency, as a float, measured in Hz

jointly.helpers.normalize(x: List[float])[source]

Normalizes signal to interval [-1, 1] with mean 0.

jointly.helpers.stretch_signals(source: pandas.core.frame.DataFrame, factor: float, start_time: pandas.core.indexes.datetimes.DatetimeIndex) pandas.core.frame.DataFrame[source]

Returns a copy of DataFrame with stretched DateTimeIndex.

Parameters
  • source – the index of this DataFrame will be stretched.

  • factor – the factor by which to streth the DateTimeIndex

  • start_time – first index, i.e., time, in the dataframe

Returns

copy of the dataframe with stretched index

jointly.helpers.verify_segments(signals: Iterable[str], segments: Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]])[source]

Verify that two synchronization points (i.e., start and end) have been found for each signal.

jointly.helpers_plotting

Contains plotting helpers

jointly.helpers_plotting.plot_reference_columns(sources: Dict[str, Dict[str, Optional[Union[str, pandas.core.frame.DataFrame, float, pandas._libs.tslibs.timedeltas.Timedelta]]]], title: str = '')[source]

Plots a normalized version of the reference columns, i.e., what jointly is detecting shakes on

Parameters
  • sources – a SourceDict

  • title – additional title if desired

jointly.helpers_plotting.plot_segments(dataframe: pandas.core.frame.DataFrame, segments: Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]], together=False, separate=False)[source]

Plot the segments of a reference signal dataframe

Parameters
  • dataframe – the dataframe

  • segments – a SyncPairs instance that specifies the data to be drawn

  • together – true to plot everything together

  • separate – true to plot separate

jointly.abstract_extractor

class jointly.abstract_extractor.AbstractExtractor[source]

Bases: object

Super class for extractor methods. First subclass is the shake extractor, which finds the location of shakes in the data.

abstract get_segments(signals: pandas.core.frame.DataFrame) Dict[str, Dict[str, Dict[str, pandas._libs.tslibs.timestamps.Timestamp]]][source]

Detect first and second segments to use for synchronization and return dictionary with start and end timestamps for each signal.

jointly.log

Sets up the logging format