resistics.time module<a class="headerlink" href="#module-resistics.time" title="Permalink to this headline">¶

copy() → resistics.time.TimeData[source]¶: Get a deepcopy of the time data object

plot(fig: Optional[plotly.graph_objs._figure.Figure] = None, chans: Optional[List[str]] = None, color: str = 'blue', legend: str = 'TimeData', max_pts: Optional[int] = 10000) → plotly.graph_objs._figure.Figure[source]¶

Plot time series data

Parameters

fig (Optional[go.Figure], optional) – A figure if appending the data to an existing plot, by default None
chans (Optional[List[str]], optional) – Explicit definition of channels to plot, by default None
color (str, optional) – The color for the data, by default “blue”
legend (str, optional) – The legend group to use, by default “TimeData”. This is more useful when plotting multiple TimeData
max_pts (Optional[int], optional) – The maximum number of points for any channel plot before applying lttbc downsampling, by default 10_000. If set to None, no downsampling will be applied.

Returns

Plotly Figure

Return type

go.Figure

Raises

ValueError – If a figure is provided and channels have not been explicitly defined

to_string() → str[source]¶: Class details as a string

pydantic model resistics.time.TimeReader[source]¶

Bases: resistics.common.ResisticsProcess

Show JSON schema

{
   "title": "TimeReader",
   "description": "Base class for resistics processes\n\nResistics processes perform operations on data (including read and write\noperations). Each time a ResisticsProcess child class is run, it should add\na process record to the dataset",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "apply_scalings": {
         "title": "Apply Scalings",
         "default": true,
         "type": "boolean"
      },
      "extension": {
         "title": "Extension",
         "type": "string"
      }
   }
}

field apply_scalings: bool = True¶

field extension: Optional[str] = None¶

run(dir_path: pathlib.Path, metadata_only: Optional[bool] = False, metadata: Optional[resistics.time.TimeMetadata] = None, from_time: Optional[Union[str, pandas._libs.tslibs.timestamps.Timestamp, datetime.datetime]] = None, to_time: Optional[Union[str, pandas._libs.tslibs.timestamps.Timestamp, datetime.datetime]] = None, from_sample: Optional[int] = None, to_sample: Optional[int] = None) → Union[resistics.time.TimeMetadata, resistics.time.TimeData][source]¶

Read time series data

Parameters

dir_path (Path) – The directory path
metadata_only (Optional[bool], optional) – Read only the metadata, by default False
metadata (Optional[TimeMetadata], optional) – Pass the metadata if its already been read in, by default None.
from_time (Union[DateTimeLike, None], optional) – Timestamp to read from, by default None
to_time (Union[DateTimeLike, None], optional) – Timestamp to read to, by default None
from_sample (Union[int, None], optional) – Sample to read from, by default None
to_sample (Union[int, None], optional) – Sample to read to, by default None

Returns

A TimeData instance

Return type

read_metadata(dir_path: pathlib.Path) → resistics.time.TimeMetadata[source]¶

Read time series data metadata

Parameters: dir_path (Path) – The directory path of the time series data
Raises: NotImplementedError – To be implemented in child classes

read_data(dir_path: pathlib.Path, metadata: resistics.time.TimeMetadata, read_from: int, read_to: int) → resistics.time.TimeData[source]¶

Read raw data with minimal scalings applied

Parameters

dir_path (path) – The directory path to read from
metadata (TimeMetadata) – Time series data metadata
read_from (int) – Sample to read data from
read_to (int) – Sample to read data to

Raises

NotImplementedError – To be implemented in child TimeReader classes

scale_data(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Scale data to physically meaningful units.

For magnetotelluric data, this is assumed to be mV/km for electric channels, mV for magnetic channels (or nT for certain sensors)

The base class assumes the data is already in the correct units and requires no scaling.

Parameters: time_data (TimeData) – TimeData read in from file
Returns: TimeData scaled to give physically meaningful units
Return type: TimeData

pydantic model resistics.time.TimeReaderJSON[source]¶

Bases: resistics.time.TimeReader

Base class for TimeReaders that use a resistics JSON header

Show JSON schema

{
   "title": "TimeReaderJSON",
   "description": "Base class for TimeReaders that use a resistics JSON header",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "apply_scalings": {
         "title": "Apply Scalings",
         "default": true,
         "type": "boolean"
      },
      "extension": {
         "title": "Extension",
         "type": "string"
      }
   }
}

read_metadata(dir_path: pathlib.Path) → resistics.time.TimeMetadata[source]¶

Read the time series data metadata and return

Parameters

dir_path (Path) – Path to time series data directory

Returns

Metadata for time series data

Return type

TimeMetadata

Raises

MetadataReadError – If the headers cannot be parsed
TimeDataReadError – If the data files do not match the expected extension

field apply_scalings: bool = True¶

field extension: Optional[str] = None¶

field name: Optional[str] [Required]¶

Validated by

validate_name

pydantic model resistics.time.TimeReaderAscii[source]¶

Bases: resistics.time.TimeReaderJSON

Class for reading Ascii data

Ascii data expected to be a single file with all the data. The delimiter can be set using the delimiter class attribute as can the number of header lines with the n_header attribute.

Show JSON schema

{
   "title": "TimeReaderAscii",
   "description": "Class for reading Ascii data\n\nAscii data expected to be a single file with all the data. The delimiter can\nbe set using the delimiter class attribute as can the number of header\nlines with the n_header attribute.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "apply_scalings": {
         "title": "Apply Scalings",
         "default": true,
         "type": "boolean"
      },
      "extension": {
         "title": "Extension",
         "default": ".txt",
         "type": "string"
      },
      "delimiter": {
         "title": "Delimiter",
         "type": "string"
      },
      "n_header": {
         "title": "N Header",
         "default": 0,
         "type": "integer"
      }
   }
}

field extension: str = '.txt'¶

field delimiter: Optional[str] = None¶

field n_header: int = 0¶

read_data(dir_path: pathlib.Path, metadata: resistics.time.TimeMetadata, read_from: int, read_to: int) → resistics.time.TimeData[source]¶

Read data from Ascii files

Parameters

dir_path (path) – The directory path to read from
metadata (TimeMetadata) – Time series data metadata
read_from (int) – Sample to read data from
read_to (int) – Sample to read data to

Returns

TimeData

Return type

Raises

ValueError – If metadata is None

pydantic model resistics.time.TimeReaderNumpy[source]¶

Bases: resistics.time.TimeReaderJSON

Class for reading Numpy data

This is expected to be a single data file for all channels. The ordering is assumed to be the same as the channels definition in the metadata.

Show JSON schema

{
   "title": "TimeReaderNumpy",
   "description": "Class for reading Numpy data\n\nThis is expected to be a single data file for all channels. The ordering is\nassumed to be the same as the channels definition in the metadata.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "apply_scalings": {
         "title": "Apply Scalings",
         "default": true,
         "type": "boolean"
      },
      "extension": {
         "title": "Extension",
         "default": ".npy",
         "type": "string"
      }
   }
}

field extension: str = '.npy'¶

read_data(dir_path: pathlib.Path, metadata: resistics.time.TimeMetadata, read_from: int, read_to: int) → resistics.time.TimeData[source]¶

Read raw data saved in numpy data

Parameters

dir_path (path) – The directory path to read from
metadata (TimeMetadata) – Time series data metadata
read_from (int) – Sample to read data from
read_to (int) – Sample to read data to

Returns

TimeData

Return type

Raises

ValueError – If metadata is None

pydantic model resistics.time.TimeWriterNumpy[source]¶

Bases: resistics.common.ResisticsWriter

Write out time data in numpy binary format

Data is written out as a single data file including all channels

Show JSON schema

{
   "title": "TimeWriterNumpy",
   "description": "Write out time data in numpy binary format\n\nData is written out as a single data file including all channels",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "overwrite": {
         "title": "Overwrite",
         "default": true,
         "type": "boolean"
      }
   }
}

run(dir_path: pathlib.Path, time_data: resistics.time.TimeData) → None[source]¶

Write out TimeData

Parameters

dir_path (Path) – The directory path to write to
time_data (TimeData) – TimeData to write out

Raises

WriteError – If unable to write to the directory

field overwrite: bool = True¶

field name: Optional[str] [Required]¶

Validated by

validate_name

pydantic model resistics.time.TimeWriterAscii[source]¶

Bases: resistics.common.ResisticsWriter

Write out time data in ascii format

Show JSON schema

{
   "title": "TimeWriterAscii",
   "description": "Write out time data in ascii format",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "overwrite": {
         "title": "Overwrite",
         "default": true,
         "type": "boolean"
      }
   }
}

run(dir_path: pathlib.Path, time_data: resistics.time.TimeData) → None[source]¶

Write out TimeData

Parameters

dir_path (Path) – The directory path to write to
time_data (TimeData) – TimeData to write out

Raises

WriteError – If unable to write to the directory

field overwrite: bool = True¶

field name: Optional[str] [Required]¶

Validated by

validate_name

resistics.time.new_time_data(time_data: resistics.time.TimeData, metadata: Optional[resistics.time.TimeMetadata] = None, data: Optional[numpy.ndarray] = None, record: Optional[resistics.common.Record] = None) → resistics.time.TimeData[source]¶

Get a new TimeData

Values are taken from an existing TimeData where they are not explicitly specified. This is useful in a process where only some aspects of the TimeData have been changed

Parameters

time_data (TimeData) – The existing TimeData
metadata (Optional[TimeMetadata], optional) – A new TimeMetadata, by default None
data (Optional[np.ndarray], optional) – New data, by default None
record (Optional[Record], optional) – A new record to add, by default None

Returns

A new TimeData instance

Return type

pydantic model resistics.time.TimeProcess[source]¶

Bases: resistics.common.ResisticsProcess

Parent class for processing time data

Show JSON schema

{
   "title": "TimeProcess",
   "description": "Parent class for processing time data",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      }
   }
}

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶: Run the time processor

field name: Optional[str] [Required]¶

Validated by

validate_name

pydantic model resistics.time.Subsection[source]¶

Get a subsection of time data

Parameters

from_time (DateTimeLike) – Time to take subsection from
to_time (DateTimeLike) – Time to take subsection to

Examples

>>> import matplotlib.pyplot as plt
>>> from resistics.testing import time_data_random
>>> from resistics.time import Subsection
>>> time_data = time_data_random(n_samples=1000)
>>> print(time_data.metadata.first_time, time_data.metadata.last_time)
2020-01-01 00:00:00 2020-01-01 00:01:39.9
>>> process = Subsection(from_time="2020-01-01 00:00:25", to_time="2020-01-01 00:00:50.9")
>>> subsection = process.run(time_data)
>>> print(subsection.metadata.first_time, subsection.metadata.last_time)
2020-01-01 00:00:25 2020-01-01 00:00:50.9
>>> subsection.metadata.n_samples
260
>>> plt.plot(time_data.get_timestamps(), time_data["Ex"], label="full") 
>>> plt.plot(subsection.get_timestamps(), subsection["Ex"], label="sub") 
>>> plt.legend(loc=3) 
>>> plt.tight_layout() 
>>> plt.show() 

Show JSON schema

{
   "title": "Subsection",
   "description": "Get a subsection of time data\n\nParameters\n----------\nfrom_time : DateTimeLike\n    Time to take subsection from\nto_time : DateTimeLike\n    Time to take subsection to\n\nExamples\n--------\n.. plot::\n    :width: 90%\n\n    >>> import matplotlib.pyplot as plt\n    >>> from resistics.testing import time_data_random\n    >>> from resistics.time import Subsection\n    >>> time_data = time_data_random(n_samples=1000)\n    >>> print(time_data.metadata.first_time, time_data.metadata.last_time)\n    2020-01-01 00:00:00 2020-01-01 00:01:39.9\n    >>> process = Subsection(from_time=\"2020-01-01 00:00:25\", to_time=\"2020-01-01 00:00:50.9\")\n    >>> subsection = process.run(time_data)\n    >>> print(subsection.metadata.first_time, subsection.metadata.last_time)\n    2020-01-01 00:00:25 2020-01-01 00:00:50.9\n    >>> subsection.metadata.n_samples\n    260\n    >>> plt.plot(time_data.get_timestamps(), time_data[\"Ex\"], label=\"full\") # doctest: +SKIP\n    >>> plt.plot(subsection.get_timestamps(), subsection[\"Ex\"], label=\"sub\") # doctest: +SKIP\n    >>> plt.legend(loc=3) # doctest: +SKIP\n    >>> plt.tight_layout() # doctest: +SKIP\n    >>> plt.show() # doctest: +SKIP",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "from_time": {
         "title": "From Time",
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "string",
               "format": "date-time"
            },
            {
               "type": "string",
               "format": "date-time"
            }
         ]
      },
      "to_time": {
         "title": "To Time",
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "string",
               "format": "date-time"
            },
            {
               "type": "string",
               "format": "date-time"
            }
         ]
      }
   },
   "required": [
      "from_time",
      "to_time"
   ]
}

field from_time: Union[str, pandas._libs.tslibs.timestamps.Timestamp, datetime.datetime] [Required]¶

field to_time: Union[str, pandas._libs.tslibs.timestamps.Timestamp, datetime.datetime] [Required]¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Take a subsection from TimeData

Parameters: time_data (TimeData) – TimeData to take subsection from
Returns: Subsection TimeData
Return type: TimeData

pydantic model resistics.time.InterpolateNans[source]¶

Interpolate nan values in the data

Preserve the data type of the input time data

Examples

>>> from resistics.testing import time_data_with_nans
>>> from resistics.time import InterpolateNans
>>> time_data = time_data_with_nans()
>>> time_data["Hx"]
array([nan,  2.,  3.,  5.,  1.,  2.,  3.,  4.,  2.,  6.,  7., nan, nan,
        4.,  3.,  2.], dtype=float32)
>>> process = InterpolateNans()
>>> time_data_new = process.run(time_data)
>>> time_data_new["Hx"]
array([2., 2., 3., 5., 1., 2., 3., 4., 2., 6., 7., 6., 5., 4., 3., 2.],
      dtype=float32)

Show JSON schema

{
   "title": "InterpolateNans",
   "description": "Interpolate nan values in the data\n\nPreserve the data type of the input time data\n\nExamples\n--------\n>>> from resistics.testing import time_data_with_nans\n>>> from resistics.time import InterpolateNans\n>>> time_data = time_data_with_nans()\n>>> time_data[\"Hx\"]\narray([nan,  2.,  3.,  5.,  1.,  2.,  3.,  4.,  2.,  6.,  7., nan, nan,\n        4.,  3.,  2.], dtype=float32)\n>>> process = InterpolateNans()\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Hx\"]\narray([2., 2., 3., 5., 1., 2., 3., 4., 2., 6., 7., 6., 5., 4., 3., 2.],\n      dtype=float32)",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      }
   }
}

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Interpolate nan values

Parameters: time_data (TimeData) – TimeData to remove nan values from
Returns: TimeData with no nan values
Return type: TimeData

field name: Optional[str] [Required]¶

Validated by

validate_name

pydantic model resistics.time.RemoveMean[source]¶

Remove channel mean value from each channel

Preserve the data type of the input time data

Examples

>>> import numpy as np
>>> from resistics.testing import time_data_simple
>>> from resistics.time import RemoveMean
>>> time_data = time_data_simple()
>>> process = RemoveMean()
>>> time_data_new = process.run(time_data)
>>> time_data_new["Hx"]
array([-2.5, -1.5, -0.5,  1.5, -2.5, -1.5, -0.5,  0.5, -1.5,  2.5,  3.5,
        2.5,  1.5,  0.5, -0.5, -1.5], dtype=float32)
>>> hx_test = time_data["Hx"] - np.mean(time_data["Hx"])
>>> hx_test
array([-2.5, -1.5, -0.5,  1.5, -2.5, -1.5, -0.5,  0.5, -1.5,  2.5,  3.5,
        2.5,  1.5,  0.5, -0.5, -1.5], dtype=float32)
>>> np.all(hx_test == time_data_new["Hx"])
True

Show JSON schema

{
   "title": "RemoveMean",
   "description": "Remove channel mean value from each channel\n\nPreserve the data type of the input time data\n\nExamples\n--------\n>>> import numpy as np\n>>> from resistics.testing import time_data_simple\n>>> from resistics.time import RemoveMean\n>>> time_data = time_data_simple()\n>>> process = RemoveMean()\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Hx\"]\narray([-2.5, -1.5, -0.5,  1.5, -2.5, -1.5, -0.5,  0.5, -1.5,  2.5,  3.5,\n        2.5,  1.5,  0.5, -0.5, -1.5], dtype=float32)\n>>> hx_test = time_data[\"Hx\"] - np.mean(time_data[\"Hx\"])\n>>> hx_test\narray([-2.5, -1.5, -0.5,  1.5, -2.5, -1.5, -0.5,  0.5, -1.5,  2.5,  3.5,\n        2.5,  1.5,  0.5, -0.5, -1.5], dtype=float32)\n>>> np.all(hx_test == time_data_new[\"Hx\"])\nTrue",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      }
   }
}

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Remove mean from TimeData

Parameters: time_data (TimeData) – TimeData input
Returns: TimeData with mean removed
Return type: TimeData

field name: Optional[str] [Required]¶

Validated by

validate_name

pydantic model resistics.time.Add[source]¶

Add values to channels

Add can be used to add a constant value to all channels or values for specific channels can be provided.

Add preserves the data type of the original data

Parameters: add (Union[float, Dict[str, float]]) – Either a scalar to add to all channels or dictionary with values to add to each channel

Examples

Using a constant value for all channels passed as a scalar

>>> from resistics.testing import time_data_ones
>>> from resistics.time import Add
>>> time_data = time_data_ones()
>>> process = Add(add=5)
>>> time_data_new = process.run(time_data)
>>> time_data_new["Ex"] - time_data["Ex"]
array([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)
>>> time_data_new["Ey"] - time_data["Ey"]
array([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)

Variable values for the channels provided as a dictionary

>>> time_data = time_data_ones()
>>> process = Add(add={"Ex": 3, "Hy": -7})
>>> time_data_new = process.run(time_data)
>>> time_data_new["Ex"] - time_data["Ex"]
array([3., 3., 3., 3., 3., 3., 3., 3., 3., 3.], dtype=float32)
>>> time_data_new["Hy"] - time_data["Hy"]
array([-7., -7., -7., -7., -7., -7., -7., -7., -7., -7.], dtype=float32)
>>> time_data_new["Ey"] - time_data["Ey"]
array([0., 0., 0., 0., 0., 0., 0., 0., 0., 0.], dtype=float32)

Show JSON schema

{
   "title": "Add",
   "description": "Add values to channels\n\nAdd can be used to add a constant value to all channels or values for\nspecific channels can be provided.\n\nAdd preserves the data type of the original data\n\nParameters\n----------\nadd : Union[float, Dict[str, float]]\n    Either a scalar to add to all channels or dictionary with values to\n    add to each channel\n\nExamples\n--------\nUsing a constant value for all channels passed as a scalar\n\n>>> from resistics.testing import time_data_ones\n>>> from resistics.time import Add\n>>> time_data = time_data_ones()\n>>> process = Add(add=5)\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Ex\"] - time_data[\"Ex\"]\narray([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)\n>>> time_data_new[\"Ey\"] - time_data[\"Ey\"]\narray([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)\n\nVariable values for the channels provided as a dictionary\n\n>>> time_data = time_data_ones()\n>>> process = Add(add={\"Ex\": 3, \"Hy\": -7})\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Ex\"] - time_data[\"Ex\"]\narray([3., 3., 3., 3., 3., 3., 3., 3., 3., 3.], dtype=float32)\n>>> time_data_new[\"Hy\"] - time_data[\"Hy\"]\narray([-7., -7., -7., -7., -7., -7., -7., -7., -7., -7.], dtype=float32)\n>>> time_data_new[\"Ey\"] - time_data[\"Ey\"]\narray([0., 0., 0., 0., 0., 0., 0., 0., 0., 0.], dtype=float32)",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "add": {
         "title": "Add",
         "anyOf": [
            {
               "type": "number"
            },
            {
               "type": "object",
               "additionalProperties": {
                  "type": "number"
               }
            }
         ]
      }
   },
   "required": [
      "add"
   ]
}

field add: Union[float, Dict[str, float]] [Required]¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Add values to the data

Parameters: time_data (TimeData) – The input TimeData
Returns: TimeData with values added
Return type: TimeData

pydantic model resistics.time.Multiply[source]¶

Multiply channels by values

Multiply can be used to add a constant value to all channels or values for specific channels can be provided.

Multiply preseves the original type of the time data

Parameters: multiplier (Union[Dict[str, float], float]) – Either a float to multiply all channels with the same value or a dictionary to specify different values for each channel

Examples

Using a constant value for all channels passed as a scalar

>>> from resistics.testing import time_data_ones
>>> from resistics.time import Multiply
>>> time_data = time_data_ones()
>>> process = Multiply(multiplier=5)
>>> time_data_new = process.run(time_data)
>>> time_data_new["Ex"]/time_data["Ex"]
array([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)
>>> time_data_new["Ey"]/time_data["Ey"]
array([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)

Variable values for the channels provided as a dictionary

>>> time_data = time_data_ones()
>>> process = Multiply(multiplier={"Ex": 3, "Hy": -7})
>>> time_data_new = process.run(time_data)
>>> time_data_new["Ex"]/time_data["Ex"]
array([3., 3., 3., 3., 3., 3., 3., 3., 3., 3.], dtype=float32)
>>> time_data_new["Hy"]/time_data["Hy"]
array([-7., -7., -7., -7., -7., -7., -7., -7., -7., -7.], dtype=float32)
>>> time_data_new["Ey"]/time_data["Ey"]
array([1., 1., 1., 1., 1., 1., 1., 1., 1., 1.], dtype=float32)

Show JSON schema

{
   "title": "Multiply",
   "description": "Multiply channels by values\n\nMultiply can be used to add a constant value to all channels or values for\nspecific channels can be provided.\n\nMultiply preseves the original type of the time data\n\nParameters\n----------\nmultiplier : Union[Dict[str, float], float]\n    Either a float to multiply all channels with the same value or a\n    dictionary to specify different values for each channel\n\nExamples\n--------\nUsing a constant value for all channels passed as a scalar\n\n>>> from resistics.testing import time_data_ones\n>>> from resistics.time import Multiply\n>>> time_data = time_data_ones()\n>>> process = Multiply(multiplier=5)\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Ex\"]/time_data[\"Ex\"]\narray([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)\n>>> time_data_new[\"Ey\"]/time_data[\"Ey\"]\narray([5., 5., 5., 5., 5., 5., 5., 5., 5., 5.], dtype=float32)\n\nVariable values for the channels provided as a dictionary\n\n>>> time_data = time_data_ones()\n>>> process = Multiply(multiplier={\"Ex\": 3, \"Hy\": -7})\n>>> time_data_new = process.run(time_data)\n>>> time_data_new[\"Ex\"]/time_data[\"Ex\"]\narray([3., 3., 3., 3., 3., 3., 3., 3., 3., 3.], dtype=float32)\n>>> time_data_new[\"Hy\"]/time_data[\"Hy\"]\narray([-7., -7., -7., -7., -7., -7., -7., -7., -7., -7.], dtype=float32)\n>>> time_data_new[\"Ey\"]/time_data[\"Ey\"]\narray([1., 1., 1., 1., 1., 1., 1., 1., 1., 1.], dtype=float32)",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "multiplier": {
         "title": "Multiplier",
         "anyOf": [
            {
               "type": "number"
            },
            {
               "type": "object",
               "additionalProperties": {
                  "type": "number"
               }
            }
         ]
      }
   },
   "required": [
      "multiplier"
   ]
}

field multiplier: Union[float, Dict[str, float]] [Required]¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Multiply the channels

Parameters: time_data (TimeData) – Input TimeData
Returns: TimeData with channels multiplied by the specified numbers
Return type: TimeData

pydantic model resistics.time.LowPass[source]¶

Apply low pass filter

Parameters

cutoff (float) – The cutoff for the low pass
order (int, optional) – Order of the filter, by default 10

Examples

Low pass to remove 20 Hz from a time series sampled at 50 Hz

import matplotlib.pyplot as plt
from resistics.testing import time_data_periodic
from resistics.time import LowPass
time_data = time_data_periodic([10, 50], fs=250, n_samples=100)
process = LowPass(cutoff=30)
filtered = process.run(time_data)
plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original")
plt.plot(filtered.get_timestamps(), filtered["chan1"], label="filtered")
plt.legend(loc=3)
plt.tight_layout()
plt.plot()

Show JSON schema

{
   "title": "LowPass",
   "description": "Apply low pass filter\n\nParameters\n----------\ncutoff : float\n    The cutoff for the low pass\norder : int, optional\n    Order of the filter, by default 10\n\nExamples\n--------\nLow pass to remove 20 Hz from a time series sampled at 50 Hz\n\n.. plot::\n    :width: 90%\n\n    import matplotlib.pyplot as plt\n    from resistics.testing import time_data_periodic\n    from resistics.time import LowPass\n    time_data = time_data_periodic([10, 50], fs=250, n_samples=100)\n    process = LowPass(cutoff=30)\n    filtered = process.run(time_data)\n    plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\")\n    plt.plot(filtered.get_timestamps(), filtered[\"chan1\"], label=\"filtered\")\n    plt.legend(loc=3)\n    plt.tight_layout()\n    plt.plot()",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "cutoff": {
         "title": "Cutoff",
         "type": "number"
      },
      "order": {
         "title": "Order",
         "default": 10,
         "type": "integer"
      }
   },
   "required": [
      "cutoff"
   ]
}

field cutoff: float [Required]¶

field order: int = 10¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Apply the low pass filter

Parameters: time_data (TimeData) – The input TimeData
Returns: The low pass filtered TimeData
Return type: TimeData
Raises: ProcessRunError – If cutoff > nyquist

pydantic model resistics.time.HighPass[source]¶

High pass filter time data

Parameters

cutoff (float) – Cutoff for the high pass filter
order (int, optional) – Order of the filter, by default 10

Examples

High pass to remove 3 Hz from signal sampled at 50 Hz

import matplotlib.pyplot as plt
from resistics.testing import time_data_periodic
from resistics.time import HighPass
time_data = time_data_periodic([10, 50], fs=250, n_samples=100)
process = HighPass(cutoff=30)
filtered = process.run(time_data)
plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original")
plt.plot(filtered.get_timestamps(), filtered["chan1"], label="filtered")
plt.legend(loc=3)
plt.tight_layout()
plt.plot()

Show JSON schema

{
   "title": "HighPass",
   "description": "High pass filter time data\n\nParameters\n----------\ncutoff : float\n    Cutoff for the high pass filter\norder : int, optional\n    Order of the filter, by default 10\n\nExamples\n--------\nHigh pass to remove 3 Hz from signal sampled at 50 Hz\n\n.. plot::\n    :width: 90%\n\n    import matplotlib.pyplot as plt\n    from resistics.testing import time_data_periodic\n    from resistics.time import HighPass\n    time_data = time_data_periodic([10, 50], fs=250, n_samples=100)\n    process = HighPass(cutoff=30)\n    filtered = process.run(time_data)\n    plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\")\n    plt.plot(filtered.get_timestamps(), filtered[\"chan1\"], label=\"filtered\")\n    plt.legend(loc=3)\n    plt.tight_layout()\n    plt.plot()",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "cutoff": {
         "title": "Cutoff",
         "type": "number"
      },
      "order": {
         "title": "Order",
         "default": 10,
         "type": "integer"
      }
   },
   "required": [
      "cutoff"
   ]
}

field cutoff: float [Required]¶

field order: int = 10¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Apply the high pass filter

Parameters: time_data (TimeData) – The input TimeData
Returns: The high pass filtered TimeData
Return type: TimeData
Raises: ProcessRunError – If cutoff > nyquist

pydantic model resistics.time.BandPass[source]¶

Band pass filter time data

Parameters

cutoff_low (float) – The low cutoff for the band pass filter
cutoff_high (float) – The high cutoff for the band pass filter
order (int, optional) – The order of the filter, by default 10

Examples

Band pass to isolate 12 Hz signal

import matplotlib.pyplot as plt
from resistics.testing import time_data_periodic
from resistics.time import BandPass
time_data = time_data_periodic([10, 50], fs=250, n_samples=100)
process = BandPass(cutoff_low=45, cutoff_high=55)
filtered = process.run(time_data)
plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original")
plt.plot(filtered.get_timestamps(), filtered["chan1"], label="filtered")
plt.legend(loc=3)
plt.tight_layout()
plt.plot()

Show JSON schema

{
   "title": "BandPass",
   "description": "Band pass filter time data\n\nParameters\n----------\ncutoff_low : float\n    The low cutoff for the band pass filter\ncutoff_high : float\n    The high cutoff for the band pass filter\norder : int, optional\n    The order of the filter, by default 10\n\nExamples\n--------\nBand pass to isolate 12 Hz signal\n\n.. plot::\n    :width: 90%\n\n    import matplotlib.pyplot as plt\n    from resistics.testing import time_data_periodic\n    from resistics.time import BandPass\n    time_data = time_data_periodic([10, 50], fs=250, n_samples=100)\n    process = BandPass(cutoff_low=45, cutoff_high=55)\n    filtered = process.run(time_data)\n    plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\")\n    plt.plot(filtered.get_timestamps(), filtered[\"chan1\"], label=\"filtered\")\n    plt.legend(loc=3)\n    plt.tight_layout()\n    plt.plot()",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "cutoff_low": {
         "title": "Cutoff Low",
         "type": "number"
      },
      "cutoff_high": {
         "title": "Cutoff High",
         "type": "number"
      },
      "order": {
         "title": "Order",
         "default": 10,
         "type": "integer"
      }
   },
   "required": [
      "cutoff_low",
      "cutoff_high"
   ]
}

field cutoff_low: float [Required]¶

field cutoff_high: float [Required]¶

field order: int = 10¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Apply the band pass filter

Parameters

time_data (TimeData) – The input TimeData

Returns

The band pass filtered TimeData

Return type

Raises

ProcessRunError – If cutoff_low > cutoff_high
ProcessRunError – If cutoff_high > nyquist

pydantic model resistics.time.Notch[source]¶

Notch filter time data

Parameters

notch (float) – The frequency to notch
band (Optional[float], optional) – The bandwidth of the filter, by default None
order (int, optional) – The order of the filter, by default 10

Examples

Notch to remove a 50 Hz signal, for example powerline noise

import matplotlib.pyplot as plt
from resistics.testing import time_data_periodic
from resistics.time import Notch
time_data = time_data_periodic([10, 50], fs=250, n_samples=100)
process = Notch(notch=50, band=10)
filtered = process.run(time_data)
plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original")
plt.plot(filtered.get_timestamps(), filtered["chan1"], label="filtered")
plt.legend(loc=3)
plt.tight_layout()
plt.plot()

Show JSON schema

{
   "title": "Notch",
   "description": "Notch filter time data\n\nParameters\n----------\nnotch : float\n    The frequency to notch\nband : Optional[float], optional\n    The bandwidth of the filter, by default None\norder : int, optional\n    The order of the filter, by default 10\n\nExamples\n--------\nNotch to remove a 50 Hz signal, for example powerline noise\n\n.. plot::\n    :width: 90%\n\n    import matplotlib.pyplot as plt\n    from resistics.testing import time_data_periodic\n    from resistics.time import Notch\n    time_data = time_data_periodic([10, 50], fs=250, n_samples=100)\n    process = Notch(notch=50, band=10)\n    filtered = process.run(time_data)\n    plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\")\n    plt.plot(filtered.get_timestamps(), filtered[\"chan1\"], label=\"filtered\")\n    plt.legend(loc=3)\n    plt.tight_layout()\n    plt.plot()",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "notch": {
         "title": "Notch",
         "type": "number"
      },
      "band": {
         "title": "Band",
         "type": "number"
      },
      "order": {
         "title": "Order",
         "default": 10,
         "type": "integer"
      }
   },
   "required": [
      "notch"
   ]
}

field notch: float [Required]¶

field band: Optional[float] = None¶

field order: int = 10¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Apply notch filter to TimeData

Parameters: time_data (TimeData) – Input TimeData
Returns: Filtered TimeData
Return type: TimeData
Raises: ProcessRunError – If notch frequency > nyquist

pydantic model resistics.time.Resample[source]¶

Resample TimeData

Note that resampling is done on np.float64 data and this will lead to a temporary increase in memory usage. Once resampling is complete, the data is converted back to its original data type.

Parameters: new_fs (int) – The new sampling frequency

Examples

Resample the data from 250 Hz to 50 Hz

>>> import matplotlib.pyplot as plt
>>> from resistics.testing import time_data_periodic
>>> from resistics.time import Resample
>>> time_data = time_data_periodic([10, 50], fs=250, n_samples=200)
>>> print(time_data.metadata.n_samples, time_data.metadata.first_time, time_data.metadata.last_time)
200 2020-01-01 00:00:00 2020-01-01 00:00:00.796
>>> process = Resample(new_fs=50)
>>> resampled = process.run(time_data)
>>> print(resampled.metadata.n_samples, resampled.metadata.first_time, resampled.metadata.last_time)
40 2020-01-01 00:00:00 2020-01-01 00:00:00.78
>>> plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original") 
>>> plt.plot(resampled.get_timestamps(), resampled["chan1"], label="resampled") 
>>> plt.legend(loc=3) 
>>> plt.tight_layout() 
>>> plt.show() 

Show JSON schema

{
   "title": "Resample",
   "description": "Resample TimeData\n\nNote that resampling is done on np.float64 data and this will lead to a\ntemporary increase in memory usage. Once resampling is complete, the data is\nconverted back to its original data type.\n\nParameters\n----------\nnew_fs : int\n    The new sampling frequency\n\nExamples\n--------\nResample the data from 250 Hz to 50 Hz\n\n.. plot::\n    :width: 90%\n\n    >>> import matplotlib.pyplot as plt\n    >>> from resistics.testing import time_data_periodic\n    >>> from resistics.time import Resample\n    >>> time_data = time_data_periodic([10, 50], fs=250, n_samples=200)\n    >>> print(time_data.metadata.n_samples, time_data.metadata.first_time, time_data.metadata.last_time)\n    200 2020-01-01 00:00:00 2020-01-01 00:00:00.796\n    >>> process = Resample(new_fs=50)\n    >>> resampled = process.run(time_data)\n    >>> print(resampled.metadata.n_samples, resampled.metadata.first_time, resampled.metadata.last_time)\n    40 2020-01-01 00:00:00 2020-01-01 00:00:00.78\n    >>> plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\") # doctest: +SKIP\n    >>> plt.plot(resampled.get_timestamps(), resampled[\"chan1\"], label=\"resampled\") # doctest: +SKIP\n    >>> plt.legend(loc=3) # doctest: +SKIP\n    >>> plt.tight_layout() # doctest: +SKIP\n    >>> plt.show() # doctest: +SKIP",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "new_fs": {
         "title": "New Fs",
         "type": "number"
      }
   },
   "required": [
      "new_fs"
   ]
}

field new_fs: float [Required]¶

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Resample TimeData

Resampling uses the polyphase method which does not assume periodicity Calculate the upsample rate and the downsampling rate and using polyphase filtering, the final sample rate is:

\[(up / down) * original sample rate\]

Therefore, to get a sampling frequency of resampFreq, want:

\[(resampFreq / sampleFreq) * sampleFreq\]

Use the fractions library to get up and down as integers which they are required to be.

Parameters: time_data (TimeData) – Input TimeData
Returns: Resampled TimeData
Return type: TimeData

pydantic model resistics.time.Decimate[source]¶

Decimate TimeData

Warning

Data is converted to np.float64 prior to decimation. This is going to cause a temporary increase in memory usage, but decimating np.float64 delivers improved results.

The decimated data is converted back to its original data type prior to being returned.

The max_factor for a single decimation step is by default set as 3. When using np.float64 data, it is possible to use a larger decimation factor, up to 13, but this does again have an impact on results.

For more information, see https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.decimate.html

Parameters: factor (int) – The decimation factor

Examples

>>> import matplotlib.pyplot as plt
>>> from resistics.testing import time_data_periodic
>>> from resistics.time import Decimate
>>> time_data = time_data_periodic([10, 50], fs=250, n_samples=200)
>>> print(time_data.metadata.n_samples, time_data.metadata.first_time, time_data.metadata.last_time)
200 2020-01-01 00:00:00 2020-01-01 00:00:00.796
>>> process = Decimate(factor=5)
>>> decimated = process.run(time_data)
>>> print(decimated.metadata.n_samples, decimated.metadata.first_time, decimated.metadata.last_time)
40 2020-01-01 00:00:00 2020-01-01 00:00:00.78
>>> plt.plot(time_data.get_timestamps(), time_data["chan1"], label="original") 
>>> plt.plot(decimated.get_timestamps(), decimated["chan1"], label="decimated") 
>>> plt.legend(loc=3) 
>>> plt.tight_layout() 
>>> plt.show() 

Show JSON schema

{
   "title": "Decimate",
   "description": "Decimate TimeData\n\n.. warning::\n\n    Data is converted to np.float64 prior to decimation. This is going to\n    cause a temporary increase in memory usage, but decimating np.float64\n    delivers improved results.\n\n    The decimated data is converted back to its original data type prior\n    to being returned.\n\n    The max_factor for a single decimation step is by default set as 3.\n    When using np.float64 data, it is possible to use a larger decimation\n    factor, up to 13, but this does again have an impact on results.\n\n    For more information, see\n    https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.decimate.html\n\nParameters\n----------\nfactor : int\n    The decimation factor\n\nExamples\n--------\n.. plot::\n    :width: 90%\n\n    >>> import matplotlib.pyplot as plt\n    >>> from resistics.testing import time_data_periodic\n    >>> from resistics.time import Decimate\n    >>> time_data = time_data_periodic([10, 50], fs=250, n_samples=200)\n    >>> print(time_data.metadata.n_samples, time_data.metadata.first_time, time_data.metadata.last_time)\n    200 2020-01-01 00:00:00 2020-01-01 00:00:00.796\n    >>> process = Decimate(factor=5)\n    >>> decimated = process.run(time_data)\n    >>> print(decimated.metadata.n_samples, decimated.metadata.first_time, decimated.metadata.last_time)\n    40 2020-01-01 00:00:00 2020-01-01 00:00:00.78\n    >>> plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], label=\"original\") # doctest: +SKIP\n    >>> plt.plot(decimated.get_timestamps(), decimated[\"chan1\"], label=\"decimated\") # doctest: +SKIP\n    >>> plt.legend(loc=3) # doctest: +SKIP\n    >>> plt.tight_layout() # doctest: +SKIP\n    >>> plt.show() # doctest: +SKIP",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "factor": {
         "title": "Factor",
         "minimum": 1,
         "type": "integer"
      },
      "max_single_factor": {
         "title": "Max Single Factor",
         "default": 3,
         "minimum": 2,
         "type": "integer"
      }
   },
   "required": [
      "factor"
   ]
}

field factor: resistics.time.ConstrainedIntValue [Required]¶

Constraints

minimum = 1

field max_single_factor: resistics.time.ConstrainedIntValue = 3¶

Constraints

minimum = 2

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Decimate TimeData

Parameters: time_data (TimeData) – Input TimeData
Returns: Decimated TimeData
Return type: TimeData

pydantic model resistics.time.ShiftTimestamps[source]¶

Shift timestamps. This method is usually used when there is an offset on the sampling, so that instead of coinciding with a second or an hour, they are offset from this.

The function interpolates the original data onto the shifted timestamps.

Parameters: shift (float) – The shift in seconds. This must be positive as data is never extrapolated

Examples

An example shifting timestamps for TimeData with a sample period of 20 seconds (fs = 1/20 = 0.05 Hz) but with an offset of 10 seconds on the timestamps

>>> from resistics.testing import time_data_with_offset
>>> from resistics.time import ShiftTimestamps
>>> time_data = time_data_with_offset(offset=10, fs=1/20, n_samples=5)
>>> [x.time().strftime('%H:%M:%S') for x in time_data.get_timestamps()]
['00:00:10', '00:00:30', '00:00:50', '00:01:10', '00:01:30']
>>> process = ShiftTimestamps(shift=10)
>>> result = process.run(time_data)
>>> [x.time().strftime('%H:%M:%S') for x in result.get_timestamps()]
['00:00:20', '00:00:40', '00:01:00', '00:01:20']
>>> plt.plot(time_data.get_timestamps(), time_data["chan1"], "bo", label="original") 
>>> plt.plot(result.get_timestamps(), result["chan1"], "rd", label="shifted") 
>>> plt.legend(loc=4) 
>>> plt.grid() 
>>> plt.tight_layout() 
>>> plt.show() 

Show JSON schema

{
   "title": "ShiftTimestamps",
   "description": "Shift timestamps. This method is usually used when there is an offset on the\nsampling, so that instead of coinciding with a second or an hour, they are\noffset from this.\n\nThe function interpolates the original data onto the shifted timestamps.\n\nParameters\n----------\nshift : float\n    The shift in seconds. This must be positive as data is never\n    extrapolated\n\nExamples\n--------\nAn example shifting timestamps for TimeData with a sample period of 20\nseconds (fs = 1/20 = 0.05 Hz) but with an offset of 10 seconds on the\ntimestamps\n\n.. plot::\n    :width: 90%\n\n    >>> from resistics.testing import time_data_with_offset\n    >>> from resistics.time import ShiftTimestamps\n    >>> time_data = time_data_with_offset(offset=10, fs=1/20, n_samples=5)\n    >>> [x.time().strftime('%H:%M:%S') for x in time_data.get_timestamps()]\n    ['00:00:10', '00:00:30', '00:00:50', '00:01:10', '00:01:30']\n    >>> process = ShiftTimestamps(shift=10)\n    >>> result = process.run(time_data)\n    >>> [x.time().strftime('%H:%M:%S') for x in result.get_timestamps()]\n    ['00:00:20', '00:00:40', '00:01:00', '00:01:20']\n    >>> plt.plot(time_data.get_timestamps(), time_data[\"chan1\"], \"bo\", label=\"original\") # doctest: +SKIP\n    >>> plt.plot(result.get_timestamps(), result[\"chan1\"], \"rd\", label=\"shifted\") # doctest: +SKIP\n    >>> plt.legend(loc=4) # doctest: +SKIP\n    >>> plt.grid() # doctest: +SKIP\n    >>> plt.tight_layout() # doctest: +SKIP\n    >>> plt.show() # doctest: +SKIP",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "shift": {
         "title": "Shift",
         "exclusiveMinimum": 0,
         "type": "number"
      }
   },
   "required": [
      "shift"
   ]
}

field shift: pydantic.types.PositiveFloat [Required]¶

Constraints

exclusiveMinimum = 0

run(time_data: resistics.time.TimeData) → resistics.time.TimeData[source]¶

Shift timestamps and interpolate data

Parameters: time_data (TimeData) – Input TimeData
Returns: TimeData with shifted timestamps and data interpolated
Return type: TimeData
Raises: ProcessRunError – If the shift is greater than the sampling frequency. This method is not supposed to be used for resampling, but simply for removing an offset from timestamps

resistics.time.serialize_custom_fnc(fnc: Callable) → str[source]¶

Serialize the custom functions

This is not really reversible and recovering parameters from ApplyFunction is not supported

Parameters: fnc (Callable) – Function to serialize
Returns: serialized output
Return type: str

pydantic model resistics.time.ApplyFunction[source]¶