Issue#7 path refactoring

environment.yml

@@ -8,8 +8,11 @@ dependencies:
   - jupyter
   - s3fs
   - pip
+  - git
+  - nepy
   - simplekml
   - pip:
     - pyubx2
     - tilemapbase
-    - dotmap
+    - dotmap
+    - git+https://github.com/emotional-cities/pluma-analysis

pluma/io/accelerometer.py

@@ -32,19 +32,11 @@ def load_accelerometer(
     Returns:
         pd.DataFrame: Dataframe with descriptive data indexed by time (Seconds)
     """
-    root = ensure_complexpath(root)
+    path = ensure_complexpath(root)
+    path.join(filename)
     try:
-        if isinstance(root, ComplexPath):
-            if root.iss3f():
-                acc_df = pd.read_csv(root.join(filename).format(),
-                                     header=None,
-                                     names=_accelerometer_header)
-            else:
-                acc_df = pd.read_csv(root.join_to_str(filename),
-                                     header=None,
-                                     names=_accelerometer_header)
-        else:
-            acc_df = pd.read_csv(root.join_to_str(filename),
+        with path.open('rb') as stream:
+            acc_df = pd.read_csv(stream,  
                                  header=None,
                                  names=_accelerometer_header)
     except FileNotFoundError:
@@ -53,6 +45,7 @@ def load_accelerometer(
     except FileExistsError:
         warnings.warn(f'Accelerometer stream file {root.join_to_str(filename)} could not be found.')
         return
+
     acc_df['Seconds'] = _HARP_T0 + pd.to_timedelta(acc_df['Seconds'].values, 's')
     acc_df['SoftwareTimestamp'] = \
         _HARP_T0 + pd.to_timedelta(acc_df['SoftwareTimestamp'].values, 's')

pluma/io/eeg.py

@@ -0,0 +1,34 @@
+import os
+import warnings
+
+import pandas as pd
+
+from nepy import easyReader, nedfReader
+
+from pluma.io.harp import _HARP_T0
+
+_eeg_header = []
+
+
+def load_eeg(
+        filename: str = 'todo.csv',
+        root: str = '') -> pd.DataFrame:
+
+    try:
+        df = pd.read_csv(
+            os.path.join(root, filename),
+            header=None,
+            names=_eeg_header)
+    except FileNotFoundError:
+        warnings.warn(f'Eeg stream file\
+            {filename} could not be found.')
+        return
+    except FileExistsError:
+        warnings.warn(f'Eeg stream file\
+            {filename} could not be found.')
+        return
+    df['Seconds'] = _HARP_T0 + pd.to_timedelta(df['Seconds'].values, 's')
+    df['SoftwareTimestamp'] = \
+        _HARP_T0 + pd.to_timedelta(df['SoftwareTimestamp'].values, 's')
+    df.set_index('Seconds', inplace=True)
+    return df

pluma/io/empatica.py

@@ -20,20 +20,11 @@ def load_empatica(filename: str = 'empatica_harp_ts.csv',
     Returns:
         DotMap: DotMap where each Empatica message type can be indexed.
     """
-    root = ensure_complexpath(root)
-
+    path = ensure_complexpath(root)
+    path.join(filename)
     try:
-        if isinstance(root, ComplexPath):
-            if root.iss3f():
-                df = pd.read_csv(root.join(filename).format(),
-                                 names=['Message', 'Seconds'],
-                                 delimiter=',', header=1)
-            else:
-                df = pd.read_csv(root.join_to_str(filename),
-                                 names=['Message', 'Seconds'],
-                                 delimiter=',', header=1)
-        else:
-            df = pd.read_csv(root.join_to_str(filename),
+        with path.open('rb') as stream:
+            df = pd.read_csv(stream,
                              names=['Message', 'Seconds'],
                              delimiter=',', header=1)
     except FileNotFoundError:
@@ -42,6 +33,7 @@ def load_empatica(filename: str = 'empatica_harp_ts.csv',
     except FileExistsError:
         warnings.warn(f'Empatica stream file {filename} could not be found.')
         return
+
     df['Seconds'] = _HARP_T0 + pd.to_timedelta(df['Seconds'].values, 's')
     df.set_index('Seconds', inplace=True)
     df['StreamId'] = df['Message'].apply(lambda x: x.split(' ')[0])

pluma/io/harp.py

@@ -35,14 +35,18 @@ def read_harp_bin(file: Union[str, ComplexPath],
     Returns:
         pd.DataFrame: Dataframe with address data indexed by time (Seconds)
     """
-    if isinstance(file, ComplexPath):
-        if file.iss3f():
-            data = np.frombuffer(file.format().read(), dtype=np.uint8)
-        else:
-            data = np.fromfile(file, dtype=np.uint8)
-    else:
-        data = np.fromfile(file, dtype=np.uint8)
-
+    path = ensure_complexpath(file)
+    try:
+        with path.open('rb') as stream:
+            data = np.frombuffer(stream.read(), dtype=np.uint8)
+    except FileNotFoundError:
+        warnings.warn(f'Harp stream file\
+            {path} could not be found.')
+        return pd.DataFrame()
+    except FileExistsError:
+        warnings.warn(f'Harp stream file\
+            {path} could not be found.')
+        return pd.DataFrame()
     if len(data) == 0:
         return None
 
@@ -81,52 +85,23 @@ def read_harp_bin(file: Union[str, ComplexPath],
             columns=['Value' + str(x) for x in np.arange(payload.shape[1])])
 
 
-def get_stream_path(streamID: int,
-                    root: Union[str, ComplexPath] = '',
-                    suffix: str = 'Streams_',
-                    ext: str = '') -> str:
-    """Helper function to generate a full path of the harp stream binary file.
-
-    Args:
-        streamID (int): Integer ID of the harp stream (aka address).
-        root (str, optional): Root path where filename is expected to be found. Defaults to ''.
-        suffix (str, optional): Suffix of the binary file name. Defaults to 'Streams_'.
-        ext (str, optional): Extension. Defaults to ''.
-
-    Returns:
-        str: The absolute path of the binary file
-    """
-    root = ensure_complexpath(root)
-    fullfile = f'{suffix}{streamID}{ext}'
-    return root.join(fullfile)
-
-
 def load_harp_stream(streamID: int,
-                     root: Union[str, ComplexPath] = '') -> pd.DataFrame:
-    """Helper function that runs read_harp_bin() with arguments built using get_stream_path()
+                     root: Union[str, ComplexPath] = '',
+                     suffix: str = 'Streams_',
+                     ext: str = '') -> pd.DataFrame:
+    """Helper function that runs read_harp_bin()
 
     Args:
         streamID (int): Integer ID of the harp stream (aka address).
-        root (str, optional): Root path where filename is expected to be found. Defaults to ''.
-
-    Raises:
-        FileExistsError: Error if a binary file is not found.
+        root (Union[str, ComplexPath], optional): Root path where filename is expected to be found. Defaults to ''.
+        suffix (str, optional): Expected file suffix. Defaults to 'Streams_'.
+        ext (str, optional): Expected file extension Defaults to ''.
 
     Returns:
         pd.DataFrame: Dataframe with address data indexed by time (Seconds)
     """
-
-    path = get_stream_path(streamID, root)
-    try:
-        return read_harp_bin(path)
-    except FileNotFoundError:
-        warnings.warn(f'Harp stream file\
-            {path} could not be found.')
-        return pd.DataFrame()
-    except FileExistsError:
-        warnings.warn(f'Harp stream file\
-            {path} could not be found.')
-        return pd.DataFrame()
-
+    path = ensure_complexpath(root)
+    path.join(f'{suffix}{streamID}{ext}')
+    return read_harp_bin(path)
 
 

pluma/io/microphone.py

@@ -1,39 +1,35 @@
-import os
 import warnings
 import numpy as np
 
 from typing import Union
 from pluma.io.path_helper import ComplexPath, ensure_complexpath
 
+
 def load_microphone(filename: str = 'Microphone.bin',
                     channels: int = 2,
                     root: Union[str, ComplexPath] = '',
-                    dtype = np.uint16
+                    dtype=np.uint16
                     ) -> np.array:
     """Loads microphone waveform data from a file into a numpy array.
 
     Args:
         filename (str, optional): Input file name to target. Defaults to 'Microphone.bin'.
         channels (int, optional): Number of expected audio input channels. Defaults to 2.
         root (str, optional): Root path where filename is expected to be found. Defaults to ''.
-
+        dtype (any, optional): Data type of the binary file. Defaults to np.uint16.
     Returns:
         np.array: Array with raw waveform data from the microphone stream.
     """
-    root = ensure_complexpath(root)
-    try:
-        if isinstance(root, ComplexPath):
-            if root.iss3f():
-                micdata = np.frombuffer(root.join(filename).format().read(), dtype=dtype)
-            else:
-                micdata = np.fromfile(root.join(filename).root, dtype=dtype)
-        else:
-            micdata = np.fromfile(os.path.join(root, filename), dtype=dtype)
+    path = ensure_complexpath(root)
+    path.join(filename)
 
-        micdata = micdata\
-            .reshape(-1, channels)
-    except FileExistsError:
-        warnings.warn(f'Microphone stream file {filename} could not be found.')
+    try:
+        with path.open('rb') as stream:
+            micdata = np.frombuffer(stream.read(), dtype=dtype)
     except FileNotFoundError:
-        warnings.warn(f'Microphone stream file {filename} could not be found.')
-    return micdata
+        warnings.warn(f'Microphone stream file\
+            {path} could not be found.')
+    except FileExistsError:
+        warnings.warn(f'Microphone stream file\
+            {path} could not be found.')
+    return micdata