Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from . import analysis
from . import io
from . import preprocessing
from .analysis import *
from .preprocessing import detect_nonwear_troiano, detect_nonwear_choi
10 changes: 10 additions & 0 deletions src/circstudio/analysis/metrics/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -692,6 +692,7 @@ def IS(data):
Returns
-------
is: float
NaN for a constant signal, where the ratio is 0/0.

Notes
-----
Expand Down Expand Up @@ -750,6 +751,10 @@ def IS(data):

d_1h = data.var()

# Constant signal: both variances are zero, so IS is undefined
if d_1h == 0:
return np.nan

return d_24h / d_1h


Expand Down Expand Up @@ -821,6 +826,7 @@ def IV(data):
Returns
-------
iv: float
NaN for a constant signal, where the ratio is 0/0.

Notes
-----
Expand Down Expand Up @@ -870,6 +876,10 @@ def IV(data):

d_1h = data.var()

# Constant signal: no variance and no successive differences, so IV is undefined
if d_1h == 0:
return np.nan

return c_1h / d_1h


Expand Down
9 changes: 8 additions & 1 deletion src/circstudio/io/atr/atr.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,14 @@ def __init__(self,
raise ValueError("The input file does not contain a header.")

#Extract information from the header
freq = pd.Timedelta(int(header['INTERVAL'][0]), unit='s')
interval_s = int(header['INTERVAL'][0]) # BUG FIX: parse interval before validating
if interval_s == 0: # BUG FIX: prevent ZeroDivisionError in pd.resample when INTERVAL==0
raise ValueError(
"INTERVAL is 0 in the ATR header — cannot create a zero-duration "
"sampling frequency. The file may be corrupt or use an unsupported "
"export configuration."
) # BUG FIX: raise a clear error instead of a cryptic ZeroDivisionError
freq = pd.Timedelta(interval_s, unit='s') # BUG FIX: build frequency from validated interval

# Create a DataFrame containing actigraphy data
data = pd.read_csv(input_fname,
Expand Down
155 changes: 125 additions & 30 deletions src/circstudio/io/mask.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,12 @@
from pandas.tseries.frequencies import to_offset

from ..analysis.tools import *
from ..preprocessing.nonwear import detect_nonwear_troiano, detect_nonwear_choi

_NONWEAR_ALGORITHMS = {
"troiano": detect_nonwear_troiano,
"choi": detect_nonwear_choi,
}


class BaseLog:
Expand Down Expand Up @@ -441,7 +447,7 @@ def create_inactivity_mask(self, duration):


def add_mask_period(self, start, stop):
"""Add a period to the inactivity mask
"""Add a period to the inactivity mask (additive — never clears existing mask).

Parameters
----------
Expand All @@ -451,42 +457,131 @@ def add_mask_period(self, start, stop):
Stop time (YYYY-MM-DD HH:MM:SS) of the inactivity period.
"""

# Check if a mask has already been created
# NB : if the inactivity_length is not None, accessing the mask will
# trigger its creation.
if self.inactivity_length is None:
self.inactivity_length = -1
# self.mask = pd.Series(
# np.ones(self.length()),
# index=self.data.index
# )

# Check if start and stop are within the index range
if pd.Timestamp(start) < self.mask.index[0]:
# If no mask exists yet, create an all-ones (keep-all) mask directly
# on self._mask without going through the inactivity_length setter —
# the setter unconditionally clears _mask, which would destroy any mask
# previously set by detect_nonwear().
if self._mask is None:
ref = (
self._original_activity
if self._original_activity is not None
else self.activity
)
self._mask = pd.Series(
np.ones(len(ref), dtype=int),
index=ref.index,
name="nonwear_mask",
)

# Validate bounds against the full mask index (not the windowed view).
if pd.Timestamp(start) < self._mask.index[0]:
raise ValueError(
(
"Attempting to set the start time of a mask period before "
+ "the actual start time of the data.\n"
+ "Mask start time: {}".format(start)
+ "Data start time: {}".format(self.mask.index[0])
)
"Attempting to set the start time of a mask period before "
"the actual start time of the data.\n"
"Mask start time: {}\n".format(start)
+ "Data start time: {}".format(self._mask.index[0])
)
if pd.Timestamp(stop) > self.mask.index[-1]:
if pd.Timestamp(stop) > self._mask.index[-1]:
raise ValueError(
(
"Attempting to set the stop time of a mask period after "
+ "the actual stop time of the data.\n"
+ "Mask stop time: {}".format(stop)
+ "Data stop time: {}".format(self.mask.index[-1])
)
"Attempting to set the stop time of a mask period after "
"the actual stop time of the data.\n"
"Mask stop time: {}\n".format(stop)
+ "Data stop time: {}".format(self._mask.index[-1])
)

# Mark the period as non-wear (0) without disturbing the rest.
self._mask = self._mask.mask(
(self._mask.index >= pd.Timestamp(start))
& (self._mask.index <= pd.Timestamp(stop)),
0,
)

def detect_nonwear(
self,
method="choi",
min_length="90min",
spike_tolerance=2,
spike_max_counts=100,
**kwargs,
):
"""Detect non-wear periods automatically and store the result as the mask.

Runs one of the validated actigraphy non-wear detection algorithms on
``self.activity`` and stores the resulting binary series as
``self._mask`` (0 = non-wear, 1 = wear), making it immediately
available via ``self.mask``. Any previously set inactivity-length
mask is discarded.

Parameters
----------
method : {"choi", "troiano"}
Algorithm to use.

* ``"choi"`` — Choi et al. (2011), *Med Sci Sports Exerc* 43(2):357-364.
Recommended default: reduces false positives during sleep by
requiring that the neighbourhood around isolated spikes is also
inactive. Default minimum run: 90 min.

* ``"troiano"`` — Troiano et al. (2008), *Med Sci Sports Exerc*
40(1):181-188. Original NHANES algorithm. Default minimum
run: 60 min.

min_length : str or int
Minimum duration of a non-wear period. Accepts a pandas offset
string (e.g. ``"60min"``, ``"2h"``) or a plain integer number of
epochs. Defaults: ``"90min"`` for Choi, ``"60min"`` for Troiano
(override here if needed).
spike_tolerance : int
Maximum number of non-zero epochs permitted within a non-wear
candidate window (provided their counts ≤ ``spike_max_counts``).
Default is 2.
spike_max_counts : int
Epochs with counts at or below this value are treated as isolated
spikes rather than genuine activity. Default is 100.
**kwargs
Additional keyword arguments forwarded to the algorithm function
(e.g. ``window_size`` for Choi).

Raises
------
ValueError
If ``method`` is not recognised.
"""
method_key = method.lower()
if method_key not in _NONWEAR_ALGORITHMS:
raise ValueError(
f"Unknown non-wear detection method '{method}'. "
f"Available methods: {list(_NONWEAR_ALGORITHMS.keys())}"
)

# Set mask values between start and stop to zeros
# self.mask.loc[start:stop] = 0
self.mask = self.mask.mask(
(self.mask.index >= start) & (self.mask.index <= stop), 0
if self.activity is None and self._original_activity is None:
warnings.warn(
"No activity data found. Cannot detect non-wear periods.",
UserWarning,
)
return

# Always run detection on the original (unfiltered) activity so that
# previously applied masks / resampling do not bias the algorithm.
base_activity = (
self._original_activity
if self._original_activity is not None
else self.activity
)

fn = _NONWEAR_ALGORITHMS[method_key]
detected_mask = fn(
activity=base_activity,
min_length=min_length,
spike_tolerance=spike_tolerance,
spike_max_counts=spike_max_counts,
**kwargs,
)

# Discard any existing length-based mask and store the new one.
self._inactivity_length = None
self._mask = detected_mask

def add_mask_periods(self, input_fname, *args, **kwargs):
"""Add periods to the inactivity mask

Expand Down
Loading