Skip to content

Commit

Permalink
Simualtion plots complete
Browse files Browse the repository at this point in the history
  • Loading branch information
daquintero committed Nov 28, 2024
1 parent 0bd0260 commit f912b3a
Show file tree
Hide file tree
Showing 46 changed files with 779 additions and 464 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -731,7 +731,7 @@ def calibration_propagation_delay_experiment(

# Note that this has some limitations of revalidation and reinstantion of python classes.

# ### Extract Software-Defined Statistics from a `DataTimeSignalData`
# ### Extract Software-Defined Statistics from a `TimeSignalData`

# `piel.analysis` also provides some functionality to analyse the corresponding time-data accordingly.
#
Expand Down
14 changes: 7 additions & 7 deletions piel/analysis/signals/time/core/compose.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,26 +2,26 @@

import numpy as np
from typing import List, Optional
from piel.types import DataTimeSignalData
from piel.types import TimeSignalData
from .off_state import (
extract_off_state_generator_from_full_state_data,
) # Adjust the import path as needed


def compose_pulses_into_signal(
pulses: List[DataTimeSignalData],
pulses: List[TimeSignalData],
baseline: float = 0.0,
noise_std: Optional[float] = None,
data_time_signal_kwargs: Optional[dict] = None,
start_time_s: Optional[float] = None,
end_time_s: Optional[float] = None,
) -> DataTimeSignalData:
) -> TimeSignalData:
"""
Composes a full signal from a list of pulses by inserting them into a continuous time array
and filling gaps with generated noise.
Parameters:
pulses (List[DataTimeSignalData]): List of pulse signals to be inserted.
pulses (List[TimeSignalData]): List of pulse signals to be inserted.
baseline (float, optional): Baseline value of the signal. Defaults to 0.0.
noise_std (float, optional): Standard deviation of the noise to be generated in gaps.
If not provided, it is estimated from the pulses.
Expand All @@ -30,7 +30,7 @@ def compose_pulses_into_signal(
end_time_s (float, optional): End time of the composed signal. If not provided, uses the last pulse's end time.
Returns:
DataTimeSignalData: The composed full signal with pulses and noise.
TimeSignalData: The composed full signal with pulses and noise.
"""
if data_time_signal_kwargs is None:
data_time_signal_kwargs = {}
Expand Down Expand Up @@ -164,8 +164,8 @@ def compose_pulses_into_signal(
# Insert the pulse data
full_data[pulse_start_idx:pulse_end_idx] = pulse_data

# Create the composed DataTimeSignalData instance
composed_signal = DataTimeSignalData(
# Create the composed TimeSignalData instance
composed_signal = TimeSignalData(
time_s=full_time_s.tolist(),
data=full_data.tolist(),
data_name="ComposedSignal",
Expand Down
12 changes: 6 additions & 6 deletions piel/analysis/signals/time/core/dimension.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,18 @@
import numpy as np
from piel.types import DataTimeSignalData, Unit
from piel.types import TimeSignalData, Unit
import logging

logger = logging.getLogger(__name__)


def resize_data_time_signal_units(
waveform: DataTimeSignalData,
waveform: TimeSignalData,
time_unit: Unit,
data_unit: Unit,
corrected_name_suffix: str = "_corrected",
) -> DataTimeSignalData:
) -> TimeSignalData:
"""
Applies unit corrections to the time and data arrays of a DataTimeSignalData object.
Applies unit corrections to the time and data arrays of a TimeSignalData object.
Parameters:
- waveform: The original waveform data.
Expand All @@ -21,7 +21,7 @@ def resize_data_time_signal_units(
- corrected_name_suffix: Suffix to append to the data name after correction.
Returns:
- A new DataTimeSignalData object with corrected time and data.
- A new TimeSignalData object with corrected time and data.
"""
# Convert time and data to NumPy arrays for efficient computation
time_array = np.array(waveform.time_s, dtype=float)
Expand Down Expand Up @@ -57,7 +57,7 @@ def resize_data_time_signal_units(
corrected_data_name = f"{waveform.data_name}{corrected_name_suffix}"

# Create and return the corrected waveform
return DataTimeSignalData(
return TimeSignalData(
time_s=corrected_time.tolist(), # Convert back to list if necessary
data=corrected_data.tolist(),
data_name=corrected_data_name,
Expand Down
18 changes: 9 additions & 9 deletions piel/analysis/signals/time/core/metrics.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import numpy as np
from piel.types import (
MultiDataTimeSignal,
MultiTimeSignalData,
ScalarMetric,
EdgeTransitionAnalysisTypes,
ScalarMetricCollection,
Expand All @@ -10,14 +10,14 @@


def extract_mean_metrics_list(
multi_data_time_signal: MultiDataTimeSignal, **kwargs
multi_data_time_signal: MultiTimeSignalData, **kwargs
) -> ScalarMetricCollection:
"""
Extracts scalar metrics from a collection of rising edge signals. Standard deviation is not calculated as this just
computes individual metrics list.
Args:
multi_data_time_signal (List[DataTimeSignalData]): A list of rising edge signals.
multi_data_time_signal (List[TimeSignalData]): A list of rising edge signals.
Returns:
ScalarMetricCollection: A collection of ScalarMetric instances containing the extracted metrics.
Expand Down Expand Up @@ -58,7 +58,7 @@ def extract_mean_metrics_list(


def extract_peak_to_peak_metrics_list(
multi_data_time_signal: MultiDataTimeSignal,
multi_data_time_signal: MultiTimeSignalData,
metric_kwargs_list: list[dict] = None,
**kwargs,
) -> ScalarMetricCollection:
Expand All @@ -67,7 +67,7 @@ def extract_peak_to_peak_metrics_list(
difference between the maximum and minimum values of the signal.
Args:
multi_data_time_signal (MultiDataTimeSignal): A collection of time signals to analyze.
multi_data_time_signal (MultiTimeSignalData): A collection of time signals to analyze.
Returns:
ScalarMetricCollection: A collection of ScalarMetric instances containing the peak-to-peak values
Expand Down Expand Up @@ -116,15 +116,15 @@ def extract_peak_to_peak_metrics_list(


def extract_multi_time_signal_statistical_metrics(
multi_data_time_signal: MultiDataTimeSignal,
multi_data_time_signal: MultiTimeSignalData,
analysis_type: EdgeTransitionAnalysisTypes = "peak_to_peak",
**kwargs,
) -> ScalarMetric:
"""
Extracts scalar metrics from a collection of rising edge signals.
Args:
multi_data_time_signal (List[DataTimeSignalData]): A list of rising edge signals.
multi_data_time_signal (List[TimeSignalData]): A list of rising edge signals.
analysis_type (piel.types.EdgeTransitionAnalysisTypes): The type of analysis to perform.
Returns:
Expand All @@ -146,15 +146,15 @@ def extract_multi_time_signal_statistical_metrics(


def extract_statistical_metrics_collection(
multi_data_time_signal: MultiDataTimeSignal,
multi_data_time_signal: MultiTimeSignalData,
analysis_types: list[EdgeTransitionAnalysisTypes],
**kwargs,
) -> ScalarMetricCollection:
"""
Extracts a collection of scalar metrics from a collection of rising edge signals based on multiple analysis types.
Args:
multi_data_time_signal (MultiDataTimeSignal): A collection of rising edge signals.
multi_data_time_signal (MultiTimeSignalData): A collection of rising edge signals.
analysis_types (list[EdgeTransitionAnalysisTypes], optional): The types of analyses to perform. Defaults to ["peak_to_peak"].
Returns:
Expand Down
40 changes: 20 additions & 20 deletions piel/analysis/signals/time/core/off_state.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import numpy as np
from typing import Callable, Optional, Dict
from piel.types import (
DataTimeSignalData,
TimeSignalData,
) # Ensure this import matches your project structure


Expand All @@ -12,7 +12,7 @@ def create_off_state_generator(
baseline: float = 0.0,
data_name: str = "off_state",
data_time_signal_kwargs: Optional[Dict] = None,
) -> Callable[[float, Optional[int]], DataTimeSignalData]:
) -> Callable[[float, Optional[int]], TimeSignalData]:
"""
Creates a generator function for the equivalent off state signal with noise.
Expand All @@ -21,18 +21,18 @@ def create_off_state_generator(
sampling_rate (float): Sampling rate in Hz.
baseline (float): Baseline signal level for the off state.
data_name (str): Name of the data signal.
data_time_signal_kwargs (dict, optional): Additional keyword arguments for DataTimeSignalData.
data_time_signal_kwargs (dict, optional): Additional keyword arguments for TimeSignalData.
Returns:
Callable[[float, Optional[int]], DataTimeSignalData]:
A function that takes duration_s (in seconds) and returns DataTimeSignalData.
Callable[[float, Optional[int]], TimeSignalData]:
A function that takes duration_s (in seconds) and returns TimeSignalData.
"""
if data_time_signal_kwargs is None:
data_time_signal_kwargs = {}

def generate_off_state(
duration_s: float, num_samples: Optional[int] = None
) -> DataTimeSignalData:
) -> TimeSignalData:
"""
Generates the off state signal data with noise for a given duration_s.
Expand All @@ -41,15 +41,15 @@ def generate_off_state(
num_samples (float): Number of samples to generate.
Returns:
DataTimeSignalData: The generated signal data.
TimeSignalData: The generated signal data.
"""
if num_samples is None:
num_samples = int(duration_s * sampling_rate)
time_s = np.linspace(0, duration_s, num_samples, endpoint=False)
noise = np.random.normal(loc=0.0, scale=noise_std, size=num_samples)
data = baseline + noise

return DataTimeSignalData(
return TimeSignalData(
time_s=time_s.tolist(),
data=data.tolist(),
data_name=data_name,
Expand All @@ -61,15 +61,15 @@ def generate_off_state(

# New function to extract parameters and create generator
def extract_off_state_generator_from_off_state_section(
off_state_data: DataTimeSignalData,
off_state_data: TimeSignalData,
data_name: Optional[str] = None,
data_time_signal_kwargs: Optional[Dict] = None,
) -> Callable[[float], DataTimeSignalData]:
) -> Callable[[float], TimeSignalData]:
"""
Extracts parameters from an existing off state DataTimeSignalData and creates a generator function.
Parameters:
off_state_data (DataTimeSignalData): The existing off state signal data.
off_state_data (TimeSignalData): The existing off state signal data.
data_name (str, optional): Name for the new data signal. Defaults to the original data_name.
data_time_signal_kwargs (dict, optional): Additional keyword arguments for DataTimeSignalData.
Expand Down Expand Up @@ -112,19 +112,19 @@ def extract_off_state_generator_from_off_state_section(


def extract_off_state_generator_from_full_state_data(
full_time_signal_data: DataTimeSignalData,
full_time_signal_data: TimeSignalData,
baseline: Optional[float] = None,
threshold: Optional[float] = None,
min_duration_s: Optional[float] = None,
sampling_rate: Optional[float] = None,
data_name: Optional[str] = None,
data_time_signal_kwargs: Optional[Dict] = None,
) -> Callable[[float, Optional[int]], DataTimeSignalData]:
) -> Callable[[float, Optional[int]], TimeSignalData]:
"""
Extracts parameters from an existing off state DataTimeSignalData and creates a generator function.
Parameters:
full_time_signal_data (DataTimeSignalData): The input signal data containing multiple states.
full_time_signal_data (TimeSignalData): The input signal data containing multiple states.
baseline (float, optional): The baseline value representing the off state.
If not provided, it is computed as the mean of the data.
threshold (float, optional): The maximum deviation from the baseline to consider as off state.
Expand Down Expand Up @@ -184,18 +184,18 @@ def extract_off_state_generator_from_full_state_data(


def extract_off_state_section(
full_time_signal_data: DataTimeSignalData,
full_time_signal_data: TimeSignalData,
baseline: Optional[float] = None,
threshold: Optional[float] = None,
min_duration_s: Optional[float] = None,
sampling_rate: Optional[float] = None,
data_time_signal_kwargs: Optional[Dict] = None,
) -> DataTimeSignalData:
) -> TimeSignalData:
"""
Extracts the off state segments from a DataTimeSignalData instance containing multiple on and off states.
Parameters:
full_time_signal_data (DataTimeSignalData): The input signal data containing multiple states.
full_time_signal_data (TimeSignalData): The input signal data containing multiple states.
baseline (float, optional): The baseline value representing the off state.
If not provided, it is computed as the mean of the data.
threshold (float, optional): The maximum deviation from the baseline to consider as off state.
Expand All @@ -206,7 +206,7 @@ def extract_off_state_section(
data_time_signal_kwargs (dict, optional): Additional keyword arguments for DataTimeSignalData.
Returns:
DataTimeSignalData: A new DataTimeSignalData instance containing only the off state segments.
TimeSignalData: A new DataTimeSignalData instance containing only the off state segments.
"""
if data_time_signal_kwargs is None:
data_time_signal_kwargs = {}
Expand Down Expand Up @@ -276,8 +276,8 @@ def extract_off_state_section(
extracted_time = np.array(extracted_time)[sorted_indices].tolist()
extracted_data = np.array(extracted_data)[sorted_indices].tolist()

# Create a new DataTimeSignalData instance
extracted_off_state = DataTimeSignalData(
# Create a new TimeSignalData instance
extracted_off_state = TimeSignalData(
time_s=extracted_time,
data=extracted_data,
data_name=full_time_signal_data.data_name + "_off_state",
Expand Down
14 changes: 7 additions & 7 deletions piel/analysis/signals/time/core/offset.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,25 @@
import numpy as np
from piel.types import DataTimeSignalData
from piel.types import TimeSignalData


def offset_to_first_rising_edge(
waveform: DataTimeSignalData,
waveform: TimeSignalData,
lower_threshold_ratio: float = 0.1,
upper_threshold_ratio: float = 0.9,
) -> DataTimeSignalData:
) -> TimeSignalData:
"""
Offsets the waveform's time axis so that the first rising edge occurs at time zero.
A rising edge is defined as the point where the signal transitions from below the lower
threshold to above the upper threshold.
Parameters:
waveform (DataTimeSignalData): The input waveform data.
waveform (TimeSignalData): The input waveform data.
lower_threshold_ratio (float): Lower threshold as a ratio of the amplitude range.
upper_threshold_ratio (float): Upper threshold as a ratio of the amplitude range.
Returns:
DataTimeSignalData: A new waveform with the time offset applied.
TimeSignalData: A new waveform with the time offset applied.
Raises:
ValueError: If no rising edge is found in the waveform.
Expand Down Expand Up @@ -68,8 +68,8 @@ def offset_to_first_rising_edge(
# Apply the offset
offset_time_array = time - offset_time

# Create a new DataTimeSignalData instance with the offset time
offset_signal = DataTimeSignalData(
# Create a new TimeSignalData instance with the offset time
offset_signal = TimeSignalData(
time_s=offset_time_array.tolist(),
data=data.tolist(),
data_name=waveform.data_name,
Expand Down
Loading

0 comments on commit f912b3a

Please sign in to comment.