# --------------------------------------------------------------------
#
# This file is part of Luna.
#
# LUNA is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Luna is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Luna. If not, see <http://www.gnu.org/licenses/>.
#
# Please see LICENSE.txt for more details.
#
# --------------------------------------------------------------------
"""Project-level Luna API.
This module exports :class:`proj`, the primary entry point for creating and
managing a Luna project/session and sample-list lifecycle.
"""
from __future__ import annotations
import lunapi.lunapi0 as _luna
import pandas as pd
try:
from IPython.display import display as _ipy_display
except ImportError:
_ipy_display = None
from .resources import resources, lp_version
from .parallel import coerce_strata as _coerce_strata
from .results import tables, cmdfile
def _coerce_var_value(value):
"""Convert Python values to Luna variable strings."""
if isinstance(value, bool):
return "1" if value else "0"
return str(value)
[docs]
class proj:
"""Manages a Luna engine session and an associated sample list.
Only one engine instance is created per Python session (singleton pattern).
Use :meth:`proj` to obtain a handle to that engine, load a sample list,
configure project-level variables, run Luna commands across all individuals
in the list, and retrieve the resulting output tables.
Examples
--------
>>> p = proj()
>>> p.sample_list('my-study.lst')
>>> results = p.proc('PSD spectrum dB sig=EEG')
>>> df = p.table('PSD', 'CH_F')
"""
# single static engine class
[docs]
eng = _luna.inaugurate()
def __init__(self, verbose = True ):
if verbose: print( "initiated lunapi",lp_version,proj.eng ,"\n" )
self.silence( False )
self.eng = _luna.inaugurate()
[docs]
def retire(self):
"""Shut down the Luna engine and release its resources.
Returns
-------
object
Status value returned by the C++ backend.
"""
return _luna.retire()
[docs]
def build( self, args ):
"""Build an internal sample list by scanning one or more folders.
Equivalent to the ``--build`` option of the Luna command-line tool.
Searches *args* for EDF and annotation files and constructs a
three-column sample list (ID, EDF path, annotation path).
See https://zzz.nyspi.org/luna/ref/helpers/#-build for details.
After a successful call, :meth:`sample_list` (with no arguments) will
return the discovered records.
Parameters
----------
args : str or list of str
One or more folder paths to scan. Optional ``--build`` flags may
also be included as list elements.
Returns
-------
object
Status value returned by the C++ backend.
"""
# first clear any existing sample list
proj.eng.clear()
# then try to build a new one
if not isinstance(args, list): args = [ args ]
return proj.eng.build_sample_list( args )
[docs]
def sample_list(self, filename = None , path = None , df = True ):
"""Read a sample list from *filename*, or return the current one.
When *filename* is given the named file is loaded as the project sample
list. When *filename* is omitted the function returns the list that is
already held in memory.
Parameters
----------
filename : str, optional
Path to a Luna sample-list file. If omitted, returns the
current in-memory sample list.
path : str, optional
If provided, sets the ``path`` project variable before reading so
that relative EDF paths in *filename* are resolved correctly.
df : bool, optional
When returning the in-memory list, wrap it in a
``pandas.DataFrame`` with columns ``['ID', 'EDF', 'Annotations']``
rather than returning the raw list. Default is ``True``.
Returns
-------
pandas.DataFrame or list
When *filename* is ``None``: the current sample list as a DataFrame
(if *df* is ``True``) or as a list of strings otherwise.
When *filename* is given: ``None`` (the count is printed to stdout).
"""
# return sample list
if filename is None:
sl = proj.eng.get_sample_list()
if df is True:
sl = pd.DataFrame( sl )
sl.columns = [ 'ID' , 'EDF', 'Annotations' ]
sl.index += 1
return sl
# set path?
if path is not None:
print( "setting path to " , path )
self.var( 'path' , path )
# read sample list from file, after clearing anything present
proj.eng.clear()
self.n = proj.eng.read_sample_list( filename )
print( "read",self.n,"individuals from" , filename )
#------------------------------------------------------------------------
[docs]
def nobs(self) -> int:
"""Return the number of individuals in the current sample list.
Returns
-------
int
Number of records in the in-memory sample list.
"""
return proj.eng.nobs()
#------------------------------------------------------------------------
[docs]
def validate( self ):
"""Validate every record in the current sample list.
Checks that each EDF file listed in the sample list exists and is
readable. Equivalent to the ``--validate`` option of the Luna
command-line tool.
See https://zzz.nyspi.org/luna/ref/helpers/#-validate for details.
Returns
-------
pandas.DataFrame
DataFrame with columns ``['ID', 'Filename', 'Valid']``, one row
per sample-list entry.
"""
tbl = proj.eng.validate_sample_list()
tbl = pd.DataFrame( tbl )
tbl.columns = [ 'ID' , 'Filename', 'Valid' ]
tbl.index += 1
return tbl
#------------------------------------------------------------------------
[docs]
def reset(self) -> None:
"""Clear the Luna problem flag so that further commands can run.
Luna sets an internal error flag when a command fails. Call this
method to clear that flag and allow subsequent evaluation to proceed.
Returns
-------
None
"""
proj.eng.reset()
[docs]
def reinit(self) -> None:
"""Re-initialize the project, clearing all results and state.
Returns
-------
None
"""
self.eng.reinit()
#------------------------------------------------------------------------
[docs]
def inst( self, n ):
"""Return an :class:`~lunapi.instance.inst` for one sample-list record.
Parameters
----------
n : int or str
When an ``int``, a **1-based** index into the sample list.
When a ``str``, the individual ID as it appears in the sample list.
Returns
-------
lunapi.instance.inst
Instance object wrapping the requested EDF record.
"""
# check bounds
if isinstance(n, int):
# use 1-based counts as inputs
n = n - 1
if n < 0 or n >= self.nobs():
raise IndexError(f"index out of bounds: sample list has {self.nobs()} record(s)")
# if the arg is a str that matches a sample-list
if isinstance(n, str):
sn = self.get_n(n)
if isinstance(sn, int): n = sn
# Lazy import avoids project<->instance import cycle at module import time.
from .instance import inst as _inst
# return based on n (from sample-list) or string/empty (new instance)
return _inst(proj.eng.inst( n ))
#------------------------------------------------------------------------
[docs]
def empty_inst( self, id, nr, rs, startdate = '01.01.00', starttime = '00.00.00' ):
"""Create a new :class:`~lunapi.instance.inst` backed by a blank EDF.
Constructs an in-memory EDF of fixed size with no signals. Signals can
be added afterwards with :meth:`~lunapi.instance.inst.insert_signal`.
Parameters
----------
id : str
Individual identifier to assign to the new record.
nr : int
Number of EDF records (data blocks).
rs : int
Duration of each EDF record in seconds.
startdate : str, optional
EDF start date in ``DD.MM.YY`` format. Default ``'01.01.00'``.
starttime : str, optional
EDF start time in ``HH.MM.SS`` format. Default ``'00.00.00'``.
Returns
-------
lunapi.instance.inst
Instance backed by the newly created blank EDF.
"""
# check inputs
nr = int( nr )
rs = int( rs )
if nr < 0:
raise ValueError('nr (number of records) must be a non-negative integer')
if rs < 0:
raise ValueError('rs (record duration in seconds) must be a non-negative integer')
# Lazy import avoids project<->instance import cycle at module import time.
from .instance import inst as _inst
# return instance of fixed size
return _inst(proj.eng.empty_inst(id, nr, rs, startdate, starttime ))
#------------------------------------------------------------------------
[docs]
def clear(self) -> None:
"""Remove all records from the current in-memory sample list.
Returns
-------
None
"""
proj.eng.clear()
#------------------------------------------------------------------------
[docs]
def silence(self, b = True , verbose = False ):
"""Suppress or restore Luna's console/log output.
Parameters
----------
b : bool
``True`` to silence output; ``False`` to re-enable it.
verbose : bool, optional
If ``True``, print a confirmation message. Default ``False``.
Returns
-------
None
"""
if verbose:
if b: print( 'silencing console outputs' )
else: print( 'enabling console outputs' )
proj.eng.silence(b)
#------------------------------------------------------------------------
[docs]
def is_silenced(self) -> bool:
"""Return whether Luna's log output is currently silenced.
Returns
-------
bool
``True`` if output is silenced, ``False`` otherwise.
"""
return proj.eng.is_silenced()
#------------------------------------------------------------------------
[docs]
def flush(self) -> None:
"""Flush the internal output buffer.
Returns
-------
None
"""
proj.eng.flush()
# --------------------------------------------------------------------------------
[docs]
def include( self, f ):
"""Load options and variables from a Luna parameter file (``@``-file).
Parameters
----------
f : str
Path to a Luna parameter file. Lines of the form ``key=value``
set project variables; lines starting with ``%`` are comments.
Returns
-------
object
Status value returned by the C++ backend.
"""
return proj.eng.include( f )
#------------------------------------------------------------------------
[docs]
def aliases( self ):
"""Display a table of signal and annotation aliases.
Prints (and returns ``None``) a DataFrame showing all alias mappings
currently registered with the engine.
Returns
-------
None
Output is displayed via ``IPython.display``.
"""
t = pd.DataFrame( proj.eng.aliases() )
t.index = t.index + 1
if len( t ) == 0: return t
t.columns = ["Type", "Preferred", "Case-insensitive, sanitized alias" ]
with pd.option_context('display.max_rows', None,):
if _ipy_display is not None:
_ipy_display(t)
#------------------------------------------------------------------------
[docs]
def var(self , key=None , value=None):
"""Get or set one or more project-level variables.
Thin alias for :meth:`vars`.
Parameters
----------
key : str, list of str, or dict, optional
Variable name(s) to get, or a ``{name: value}`` dict to set.
value : str, optional
Value to assign when *key* is a single variable name.
Returns
-------
str, dict, or None
The variable value (or dict of values) when getting;
``None`` when setting.
"""
return self.vars( key, value )
#------------------------------------------------------------------------
[docs]
def vars(self , key=None , value=None):
"""Get or set one or more project-level variables.
When called with no arguments, returns all currently set variables.
When *key* is a string and *value* is omitted, returns that variable's
value. When *key* is a list, returns a dict of values. When both
*key* and *value* are provided (or *key* is a dict), sets the
variable(s).
Parameters
----------
key : str, list of str, or dict, optional
Variable name(s) to get, or a ``{name: value}`` dict to set.
value : str, optional
Value to assign when *key* is a single variable name string.
Returns
-------
str, dict, or None
The variable value (or dict of values) when getting;
``None`` when setting.
"""
# return all vars?
if key is None:
return proj.eng.get_all_opts()
# return one or more vars?
if value is None:
# return 1?
if isinstance(key, str):
return proj.eng.get_opt( key )
# return some?
if isinstance(key, list):
return proj.eng.get_opts( key )
# set from a dict
if isinstance(key, dict):
for k, v in key.items():
self.vars(k,v)
return
# set a single pair
proj.eng.opt( key, _coerce_var_value( value ) )
#------------------------------------------------------------------------
[docs]
def clear_vars(self,key = None ):
"""Clear one, several, or all project-level variables.
Parameters
----------
key : str or list of str, optional
Name(s) of the variable(s) to remove. If omitted, **all**
project variables are cleared (including the ``sig`` channel
selection list).
Returns
-------
None
"""
# clear all
if key is None:
proj.eng.clear_all_opts()
# and a special case: the sig list
self.vars( 'sig', '' )
return
# clear some/one
if not isinstance(key, list): key = [ key ]
proj.eng.clear_opts(key)
#------------------------------------------------------------------------
[docs]
def clear_ivars(self):
"""Clear individual-level variables for every individual in the sample list.
Returns
-------
None
"""
proj.eng.clear_ivars()
#------------------------------------------------------------------------
[docs]
def get_n(self,id):
"""Return the 0-based internal index for a given individual ID.
Parameters
----------
id : str
Individual identifier as it appears in the sample list.
Returns
-------
int
0-based index of *id* within the sample list, or ``None`` if
not found.
"""
return proj.eng.get_n(id)
#------------------------------------------------------------------------
[docs]
def get_id(self,n):
"""Return the individual ID for a given (0-based) sample-list index.
Parameters
----------
n : int
0-based position in the sample list.
Returns
-------
str
Individual identifier at position *n*.
"""
return proj.eng.get_id(n)
#------------------------------------------------------------------------
[docs]
def get_edf(self,x):
"""Return the EDF file path for an individual.
Parameters
----------
x : int or str
Either a 0-based integer index or the individual's string ID.
Returns
-------
str
Absolute (or sample-list-relative) path to the EDF file.
"""
if ( isinstance(x,int) ):
return proj.eng.get_edf(x)
else:
return proj.eng.get_edf(proj.eng.get_n(x))
#------------------------------------------------------------------------
[docs]
def get_annots(self,x):
"""Return the annotation file path(s) for an individual.
Parameters
----------
x : int or str
Either a 0-based integer index or the individual's string ID.
Returns
-------
str
Annotation file path(s) as stored in the sample list
(comma-separated if multiple files are listed).
"""
if ( isinstance(x,int) ):
return proj.eng.get_annot(x)
else:
return proj.eng.get_annot(proj.eng.get_n(x))
#------------------------------------------------------------------------
[docs]
def import_db(self,f,s=None):
"""Import a Luna *destrat*-style output database into the result store.
Parameters
----------
f : str
Path to a Luna output database file (``*.db``).
s : str or list of str, optional
If provided, import only the individuals whose IDs match *s*.
Returns
-------
object
Status value returned by the C++ backend.
"""
if s is None:
return proj.eng.import_db(f)
else:
return proj.eng.import_db_subset(f,s)
#------------------------------------------------------------------------
[docs]
def desc( self ):
"""Display a summary table of all sample-list individuals.
Runs the Luna ``DESC`` command silently across the sample list and
displays a DataFrame with columns:
``['ID', 'Gapped', 'Date', 'Start(hms)', 'Stop(hms)', 'Dur(hms)',
'Dur(s)', '# sigs', '# annots', 'Signals']``.
Returns
-------
None
Output is displayed via ``IPython.display``.
"""
silence_mode = self.is_silenced()
self.silence(True,False)
t = pd.DataFrame( proj.eng.desc() )
self.silence( silence_mode , False )
t.index = t.index + 1
if len( t ) == 0: return t
t.columns = ["ID","Gapped","Date","Start(hms)","Stop(hms)","Dur(hms)","Dur(s)","# sigs","# annots","Signals" ]
with pd.option_context('max_colwidth',None):
if _ipy_display is not None:
_ipy_display(t)
#------------------------------------------------------------------------
[docs]
def proc(self, cmdstr ) -> ProcResult:
"""Evaluate one or more Luna commands across all sample-list individuals.
Results are stored internally and also returned. Use :meth:`table` or
:meth:`strata` to interrogate specific outputs afterwards.
Parameters
----------
cmdstr : str
One or more Luna commands, optionally separated by newlines.
Returns
-------
ProcResult
"""
from .parallel import ProcResult, _errors_frame, _stdout_frame, _records_frame
proj.eng.eval(cmdstr)
return ProcResult(
_owner=self,
errors=_errors_frame([]),
stdout=_stdout_frame([]),
records=_records_frame([]),
workers=1,
)
#------------------------------------------------------------------------
[docs]
def proc_parallel(self, cmdstr, workers=None, batch_size=None, params=None,
param_file=None, strict=False, progress=True,
out_db=None, out_text=None, in_memory=None,
n1=None, n2=None, ids=None, skip=None):
"""Evaluate Luna commands across the sample list using worker processes.
This is intended for file-backed project sample lists. Each worker
process initializes its own Luna engine and receives only the explicit
parameters passed via *params* or *param_file*; project variables from
the parent process are not implicitly inherited.
Parameters
----------
cmdstr : str
One or more Luna commands, optionally separated by newlines.
workers : int, optional
Number of worker processes. Defaults to half the available CPUs,
capped at 10 and at the number of sample-list records.
batch_size : int, optional
Number of records per submitted worker batch. Smaller values provide
more frequent progress updates. Defaults to one batch per worker,
or one record per batch when ``progress=True``.
params : dict or iterable of (str, object), optional
Project-level Luna variables to apply in each worker before each
record is evaluated. Values override duplicate keys from
*param_file*.
param_file : str or path-like, optional
Luna-style parameter file. Blank lines and ``%`` comments are
skipped; tab, space, and ``=`` delimiters are supported.
strict : bool, optional
If ``True``, raise ``ParallelProcError`` when any record or worker
fails. The exception contains the partial result.
progress : bool or callable, optional
If ``True``, show a tqdm progress bar. If callable, receive simple
parent-process progress event dicts.
n1 : int, optional
First sample-list row to process (1-based, inclusive). Mirrors
Luna's ``n1`` option.
n2 : int, optional
Last sample-list row to process (1-based, inclusive). Mirrors
Luna's ``n2`` option.
ids : str or list of str, optional
Process only individuals whose ID appears in this list. A plain
string is split on whitespace. Mirrors Luna's ``id=`` option.
skip : str or list of str, optional
Exclude individuals whose ID appears in this list. A plain
string is split on whitespace. Mirrors Luna's ``skip=`` option.
Returns
-------
lunapi.parallel.ParallelProcResult
Object containing concatenated ``tables`` plus ``errors``,
``stdout`` and per-record metadata.
"""
from .parallel import run_parallel_project
return run_parallel_project(
self,
cmdstr,
workers=workers,
batch_size=batch_size,
params=params,
param_file=param_file,
strict=strict,
progress=progress,
out_db=out_db,
out_text=out_text,
in_memory=in_memory,
n1=n1,
n2=n2,
ids=ids,
skip=skip,
)
#------------------------------------------------------------------------
[docs]
def procn(self, cmdstr, workers=None, batch_size=None, params=None,
param_file=None, strict=False, progress=True,
out_db=None, out_text=None, in_memory=None,
n1=None, n2=None, ids=None, skip=None):
"""Evaluate Luna commands across the sample list using N worker processes.
Convenience alias for :meth:`proc_parallel`.
"""
kwargs = dict(
workers=workers,
batch_size=batch_size,
params=params,
param_file=param_file,
strict=strict,
progress=progress,
)
optional = {
"out_db": out_db,
"out_text": out_text,
"in_memory": in_memory,
"n1": n1,
"n2": n2,
"ids": ids,
"skip": skip,
}
kwargs.update({key: value for key, value in optional.items() if value is not None})
return self.proc_parallel(cmdstr, **kwargs)
#------------------------------------------------------------------------
[docs]
def silent_proc(self, cmdstr ) -> ProcResult:
"""Evaluate Luna commands across all sample-list individuals without printing log output.
Identical to :meth:`proc` but suppresses console output for the
duration of the call, then restores the previous silence state.
Parameters
----------
cmdstr : str
One or more Luna commands, optionally separated by newlines.
Returns
-------
ProcResult
"""
from .parallel import ProcResult, _errors_frame, _stdout_frame, _records_frame
silence_mode = self.is_silenced()
self.silence(True,False)
proj.eng.eval(cmdstr)
self.silence( silence_mode , False )
return ProcResult(
_owner=self,
errors=_errors_frame([]),
stdout=_stdout_frame([]),
records=_records_frame([]),
workers=1,
)
#------------------------------------------------------------------------
[docs]
def commands( self ):
"""Return a DataFrame listing the commands present in the output store.
Returns
-------
pandas.DataFrame
Single-column DataFrame with column ``'Command'``.
"""
t = pd.DataFrame( proj.eng.commands() )
t.columns = ["Command"]
return t
#------------------------------------------------------------------------
[docs]
def empty_result_set( self ) -> bool:
"""Return ``True`` if the result store contains no output tables.
Returns
-------
bool
``True`` when no results are stored; ``False`` otherwise.
"""
return len( proj.eng.strata() ) == 0
#------------------------------------------------------------------------
[docs]
def strata( self ):
"""Return a DataFrame of command/strata pairs from the output store.
Returns
-------
pandas.DataFrame or None
DataFrame with columns ``['Command', 'Strata']``, or ``None`` if
the result store is empty.
"""
if self.empty_result_set(): return None
t = pd.DataFrame( proj.eng.strata() )
t.columns = ["Command","Strata"]
return t
#------------------------------------------------------------------------
[docs]
def table( self, cmd , strata = 'BL' ):
"""Return a specific output table as a DataFrame.
Parameters
----------
cmd : str
Luna command name (e.g. ``'PSD'``, ``'STAGE'``).
strata : str, optional
Stratum label for the desired table (e.g. ``'CH_F'``, ``'E'``).
Defaults to ``'BL'`` (baseline / un-stratified).
Returns
-------
pandas.DataFrame or None
Result table, or ``None`` if the result store is empty.
"""
if self.empty_result_set(): return None
strata = _coerce_strata( proj.eng.strata(), cmd, strata )
r = proj.eng.table( cmd , strata )
t = pd.DataFrame( r[1] ).T
t.columns = r[0]
return t
#------------------------------------------------------------------------
[docs]
def variables( self, cmd , strata = 'BL' ):
"""Return the variable names present in a specific output table.
Parameters
----------
cmd : str
Luna command name.
strata : str, optional
Stratum label. Defaults to ``'BL'``.
Returns
-------
list of str or None
Variable (column) names for the requested table, or ``None`` if
the result store is empty.
"""
if self.empty_result_set(): return None
return proj.eng.vars( cmd , strata )
#
# --------------------------------------------------------------------------------
# project level wrapper functions
#
# --------------------------------------------------------------------------------
[docs]
def pops( self, s = None, s1 = None , s2 = None,
path = None , lib = None ,
do_edger = True ,
no_filter = False ,
do_reref = False ,
m = None , m1 = None , m2 = None,
lights_off = '.' , lights_on = '.' ,
ignore_obs = False,
args = '' ):
"""Run the POPS automatic sleep stager across the whole sample list.
POPS (Population-based sleep staging) uses a pre-trained model to
assign sleep stages to each epoch. Results are returned as a
DataFrame from the ``POPS`` command.
Call this method in **single-channel** mode by setting *s* only, or
in **two-channel** mode by setting *s1* and *s2*.
Parameters
----------
s : str, optional
EEG channel label for single-channel staging.
s1 : str, optional
First EEG channel for two-channel staging.
s2 : str, optional
Second EEG channel for two-channel staging.
path : str, optional
Path to the POPS resource folder. Defaults to
``resources.POPS_PATH``.
lib : str, optional
POPS library name (sub-folder within *path*). Defaults to
``resources.POPS_LIB`` (``'s2'``).
do_edger : bool, optional
Apply EDGER artifact detection. Default ``True``.
no_filter : bool, optional
Skip bandpass pre-filtering of the EEG. Default ``False``.
do_reref : bool, optional
Re-reference the EEG to a mastoid channel before staging.
Default ``False``.
m : str, optional
Mastoid channel for single-channel re-referencing (required when
*do_reref* is ``True`` and *s* is set).
m1 : str, optional
First mastoid channel for two-channel re-referencing.
m2 : str, optional
Second mastoid channel for two-channel re-referencing.
lights_off : str, optional
Lights-off time as ``'HH:MM:SS'``. Use ``'.'`` if unknown.
Default ``'.'``.
lights_on : str, optional
Lights-on time as ``'HH:MM:SS'``. Use ``'.'`` if unknown.
Default ``'.'``.
ignore_obs : bool, optional
Ignore any observed (manual) staging annotations.
Default ``False``.
args : str, optional
Additional options to append to the ``POPS`` Luna command string.
Default ``''``.
Returns
-------
pandas.DataFrame or str
DataFrame of per-epoch staging results from the ``POPS`` command,
or an error string if the resource path cannot be opened.
"""
if path is None: path = resources.POPS_PATH
if lib is None: lib = resources.POPS_LIB
if not os.path.isdir( path ):
raise FileNotFoundError(f'POPS resource path not found: {path}')
if s is None and s1 is None:
raise ValueError('must set s or s1 and s2 to EEGs')
if ( s1 is None ) != ( s2 is None ):
raise ValueError('must set s or s1 and s2 to EEGs')
# POPS templates may reference mastoid vars even when do_reref is false.
# Provide safe placeholders unless rereferencing is explicitly requested.
if do_reref:
if s is not None and m is None:
raise ValueError('must set m to the mastoid channel when do_reref=True in single-channel mode')
if s1 is not None and ( m1 is None or m2 is None ):
raise ValueError('must set m1 and m2 to mastoid channels when do_reref=True in two-channel mode')
else:
if m is None: m = '.'
if m1 is None: m1 = '.'
if m2 is None: m2 = '.'
# set options
self.var( 'mpath' , path )
self.var( 'lib' , lib )
self.var( 'do_edger' , '1' if do_edger else '0' )
self.var( 'do_reref' , '1' if do_reref else '0' )
self.var( 'no_filter' , '1' if no_filter else '0' )
self.var( 'LOFF' , lights_off )
self.var( 'LON' , lights_on )
if s is not None: self.var( 's' , s )
else: self.clear_vars( 's' )
if m is not None: self.var( 'm' , m )
else: self.clear_vars( 'm' )
if s1 is not None: self.var( 's1' , s1 )
else: self.clear_vars( 's1' )
if s2 is not None: self.var( 's2' , s2 )
else: self.clear_vars( 's2' )
if m1 is not None: self.var( 'm1' , m1 )
else: self.clear_vars( 'm1' )
if m2 is not None: self.var( 'm2' , m2 )
else: self.clear_vars( 'm2' )
# get either one- or two-channel mode Luna script from POPS folder
twoch = s1 is not None and s2 is not None;
if twoch: cmdstr = cmdfile( path + '/s2.ch2.txt' )
else: cmdstr = cmdfile( path + '/s2.ch1.txt' )
# swap in any additional options to POPS
if ignore_obs is True:
args = args + ' ignore-obs-staging';
if do_edger is True:
cmdstr = cmdstr.replace( 'EDGER' , 'EDGER all' )
if args != '':
cmdstr = cmdstr.replace( 'POPS' , 'POPS ' + args + ' ')
# run the command
self.proc( cmdstr )
# return of results
return self.table( 'POPS' )
# --------------------------------------------------------------------------------
[docs]
def predict_SUN2019( self, cen , th = '3' , path = None ):
"""Run the SUN2019 brain-age prediction model for the whole sample list.
Applies the SUN2019 EEG-based brain-age prediction model to every
individual in the sample list. The individual ``${age}`` variable
must be set via a vars file before calling this method, e.g.::
proj.var('vars', 'ages.txt')
Parameters
----------
cen : str or list of str
EEG centroid channel label(s). A list is joined with commas
before being passed to Luna.
th : str or int, optional
Outlier threshold (in standard deviations) for feature exclusion.
Default ``'3'``.
path : str, optional
Path to the Luna models folder. Defaults to
``resources.MODEL_PATH``.
Returns
-------
pandas.DataFrame
DataFrame of prediction results from the ``PREDICT`` command.
"""
if path is None: path = resources.MODEL_PATH
if isinstance(cen, list): cen = ','.join( cen )
self.var( 'cen' , cen )
self.var( 'mpath' , path )
self.var( 'th' , str(th) )
self.proc( cmdfile( resources.MODEL_PATH + '/m1-adult-age-luna.txt' ) )
return self.table( 'PREDICT' )
# ================================================================================
# --------------------------------------------------------------------------------
#
# inst class
#
# --------------------------------------------------------------------------------
# ================================================================================
__all__ = ["proj"]