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, seeResultTableSpec
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
andref_column
.data
is a pandasDataFrame
with aDateTimeIndex
.ref_column
specifies the column withindata
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
andsecond
.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
andend
.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
jointly.helpers
Contains plotting helpers
- 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
- 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]
- 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.
- 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.abstract_extractor
jointly.log
Sets up the logging format