resistics.common module

Common resistics functions and classes used throughout the package

resistics.common.get_version() → str[source]

Get the version of resistics

resistics.common.is_file(file_path: pathlib.Path) → bool[source]

Check if a path exists and points to a file

Parameters

file_path (Path) – The path to check

Returns

True if it exists and is a file, False otherwise

Return type

bool

resistics.common.assert_file(file_path: pathlib.Path) → None[source]

Require that a file exists

Parameters

file_path (Path) – The path to check

Raises

  • FileNotFoundError – If the path does not exist

  • NotFileError – If the path is not a file

resistics.common.is_dir(dir_path: pathlib.Path) → bool[source]

Check if a path exists and points to a directory

Parameters

dir_path (Path) – The path to check

Returns

True if it exists and is a directory, False otherwise

Return type

bool

resistics.common.assert_dir(dir_path: pathlib.Path) → None[source]

Require that a path is a directory

Parameters

dir_path (Path) – Path to check

Raises

  • FileNotFoundError – If the path does not exist

  • NotDirectoryError – If the path is not a directory

resistics.common.dir_contents(dir_path: pathlib.Path) → Tuple[List[pathlib.Path], List[pathlib.Path]][source]

Get contents of directory

Includes both files and directories

Parameters

dir_path (Path) – Parent directory path

Returns

  • dirs (list) – List of directories

  • files (list) – List of files excluding hidden files

Raises
resistics.common.dir_files(dir_path: pathlib.Path) → List[pathlib.Path][source]

Get files in directory

Excludes hidden files

Parameters

dir_path (Path) – Parent directory path

Returns

files – List of files excluding hidden files

Return type

list

resistics.common.dir_subdirs(dir_path: pathlib.Path) → List[pathlib.Path][source]

Get subdirectories in directory

Excludes hidden files

Parameters

dir_path (Path) – Parent directory path

Returns

dirs – List of subdirectories

Return type

list

resistics.common.is_electric(chan: str) → bool[source]

Check if a channel is electric

Parameters

chan (str) – Channel name

Returns

True if channel is electric

Return type

bool

Examples

>>> from resistics.common import is_electric
>>> is_electric("Ex")
True
>>> is_electric("Hx")
False
resistics.common.is_magnetic(chan: str) → bool[source]

Check if channel is magnetic

Parameters

chan (str) – Channel name

Returns

True if channel is magnetic

Return type

bool

Examples

>>> from resistics.common import is_magnetic
>>> is_magnetic("Ex")
False
>>> is_magnetic("Hx")
True
resistics.common.get_chan_type(chan: str) → str[source]

Get the channel type from the channel name

Parameters

chan (str) – The name of the channel

Returns

The channel type

Return type

str

Raises

ValueError – If the channel is not known to resistics

Examples

>>> from resistics.common import get_chan_type
>>> get_chan_type("Ex")
'electric'
>>> get_chan_type("Hz")
'magnetic'
>>> get_chan_type("abc")
Traceback (most recent call last):
...
ValueError: Channel abc not recognised as either electric or magnetic
resistics.common.check_chan(chan: str, chans: Collection[str]) → bool[source]

Check a channel exists and raise a KeyError if not

Parameters

  • chan (str) – The channel to check

  • chans (Collection[str]) – A collection of channels to check against

Returns

True if all checks passed

Return type

bool

Raises

ChannelNotFoundError – If the channel is not found in the channel list

resistics.common.fs_to_string(fs: float) → str[source]

Convert sampling frequency into a string for filenames

Parameters

fs (float) – The sampling frequency

Returns

Sample frequency converted to string for the purposes of a filename

Return type

str

Examples

>>> from resistics.common import fs_to_string
>>> fs_to_string(512.0)
'512_000000'
resistics.common.array_to_string(data: numpy.ndarray, sep: str = ', ', precision: int = 8, scientific: bool = False) → str[source]

Convert an array to a string for logging or printing

Parameters

  • data (np.ndarray) – The array

  • sep (str, optional) – The separator to use, by default “, “

  • precision (int, optional) – Number of decimal places, by default 8. Ignored for integers.

  • scientific (bool, optional) – Flag for formatting floats as scientific, by default False

Returns

String representation of array

Return type

str

Examples

>>> import numpy as np
>>> from resistics.common import array_to_string
>>> data = np.array([1,2,3,4,5])
>>> array_to_string(data)
'1, 2, 3, 4, 5'
>>> data = np.array([1,2,3,4,5], dtype=np.float32)
>>> array_to_string(data)
'1.00000000, 2.00000000, 3.00000000, 4.00000000, 5.00000000'
>>> array_to_string(data, precision=3, scientific=True)
'1.000e+00, 2.000e+00, 3.000e+00, 4.000e+00, 5.000e+00'
pydantic model resistics.common.ResisticsModel[source]

Bases: pydantic.main.BaseModel

Base resistics model

Show JSON schema
{
   "title": "ResisticsModel",
   "description": "Base resistics model",
   "type": "object",
   "properties": {}
}

to_string() → str[source]

Class info as string

summary() → None[source]

Print a summary of the class

pydantic model resistics.common.ResisticsFile[source]

Bases: resistics.common.ResisticsModel

Required information for writing out a resistics file

Show JSON schema
{
   "title": "ResisticsFile",
   "description": "Required information for writing out a resistics file",
   "type": "object",
   "properties": {
      "created_on_local": {
         "title": "Created On Local",
         "type": "string",
         "format": "date-time"
      },
      "created_on_utc": {
         "title": "Created On Utc",
         "type": "string",
         "format": "date-time"
      },
      "version": {
         "title": "Version",
         "type": "string"
      }
   }
}

field created_on_local: datetime.datetime [Required]
field created_on_utc: datetime.datetime [Required]
field version: Optional[str] [Required]
pydantic model resistics.common.Metadata[source]

Bases: resistics.common.ResisticsModel

Parent class for metadata

Show JSON schema
{
   "title": "Metadata",
   "description": "Parent class for metadata",
   "type": "object",
   "properties": {}
}

pydantic model resistics.common.WriteableMetadata[source]

Bases: resistics.common.Metadata

Base class for writeable metadata

Show JSON schema
{
   "title": "WriteableMetadata",
   "description": "Base class for writeable metadata",
   "type": "object",
   "properties": {
      "file_info": {
         "$ref": "#/definitions/ResisticsFile"
      }
   },
   "definitions": {
      "ResisticsFile": {
         "title": "ResisticsFile",
         "description": "Required information for writing out a resistics file",
         "type": "object",
         "properties": {
            "created_on_local": {
               "title": "Created On Local",
               "type": "string",
               "format": "date-time"
            },
            "created_on_utc": {
               "title": "Created On Utc",
               "type": "string",
               "format": "date-time"
            },
            "version": {
               "title": "Version",
               "type": "string"
            }
         }
      }
   }
}

field file_info: Optional[resistics.common.ResisticsFile] = None

Information about a file, relevant if writing out or reading back in

write(json_path: pathlib.Path)[source]

Write out JSON metadata file

Parameters

json_path (Path) – Path to write JSON file

pydantic model resistics.common.Record[source]

Bases: resistics.common.ResisticsModel

Class to hold a record

A record holds information about a process that was run. It is intended to track processes applied to data, allowing a process history to be saved along with any datasets.

Examples

A simple example of creating a process record

>>> from resistics.common import Record
>>> messages = ["message 1", "message 2"]
>>> record = Record(
...     creator={"name": "example", "parameter1": 15},
...     messages=messages,
...     record_type="example"
... )
>>> record.summary()
{
    'time_local': '...',
    'time_utc': '...',
    'creator': {'name': 'example', 'parameter1': 15},
    'messages': ['message 1', 'message 2'],
    'record_type': 'example'
}

Show JSON schema
{
   "title": "Record",
   "description": "Class to hold a record\n\nA record holds information about a process that was run. It is intended to\ntrack processes applied to data, allowing a process history to be saved\nalong with any datasets.\n\nExamples\n--------\nA simple example of creating a process record\n\n>>> from resistics.common import Record\n>>> messages = [\"message 1\", \"message 2\"]\n>>> record = Record(\n...     creator={\"name\": \"example\", \"parameter1\": 15},\n...     messages=messages,\n...     record_type=\"example\"\n... )\n>>> record.summary()\n{\n    'time_local': '...',\n    'time_utc': '...',\n    'creator': {'name': 'example', 'parameter1': 15},\n    'messages': ['message 1', 'message 2'],\n    'record_type': 'example'\n}",
   "type": "object",
   "properties": {
      "time_local": {
         "title": "Time Local",
         "type": "string",
         "format": "date-time"
      },
      "time_utc": {
         "title": "Time Utc",
         "type": "string",
         "format": "date-time"
      },
      "creator": {
         "title": "Creator",
         "type": "object"
      },
      "messages": {
         "title": "Messages",
         "type": "array",
         "items": {
            "type": "string"
         }
      },
      "record_type": {
         "title": "Record Type",
         "type": "string"
      }
   },
   "required": [
      "creator",
      "messages",
      "record_type"
   ]
}

field time_local: datetime.datetime [Required]

The local time when the process ran

field time_utc: datetime.datetime [Required]

The UTC time when the process ran

field creator: Dict[str, Any] [Required]

The creator and its parameters as a dictionary

field messages: List[str] [Required]

Any messages in the record

field record_type: str [Required]

The record type

pydantic model resistics.common.History[source]

Bases: resistics.common.ResisticsModel

Class for storing processing history

Parameters

records (List[Record], optional) – List of records, by default []

Examples

>>> from resistics.testing import record_example1, record_example2
>>> from resistics.common import History
>>> record1 = record_example1()
>>> record2 = record_example2()
>>> history = History(records=[record1, record2])
>>> history.summary()
{
    'records': [
        {
            'time_local': '...',
            'time_utc': '...',
            'creator': {
                'name': 'example1',
                'a': 5,
                'b': -7.0
            },
            'messages': ['Message 1', 'Message 2'],
            'record_type': 'process'
        },
        {
            'time_local': '...',
            'time_utc': '...',
            'creator': {
                'name': 'example2',
                'a': 'parzen',
                'b': -21
            },
            'messages': ['Message 5', 'Message 6'],
            'record_type': 'process'
        }
    ]
}

Show JSON schema
{
   "title": "History",
   "description": "Class for storing processing history\n\nParameters\n----------\nrecords : List[Record], optional\n    List of records, by default []\n\nExamples\n--------\n>>> from resistics.testing import record_example1, record_example2\n>>> from resistics.common import History\n>>> record1 = record_example1()\n>>> record2 = record_example2()\n>>> history = History(records=[record1, record2])\n>>> history.summary()\n{\n    'records': [\n        {\n            'time_local': '...',\n            'time_utc': '...',\n            'creator': {\n                'name': 'example1',\n                'a': 5,\n                'b': -7.0\n            },\n            'messages': ['Message 1', 'Message 2'],\n            'record_type': 'process'\n        },\n        {\n            'time_local': '...',\n            'time_utc': '...',\n            'creator': {\n                'name': 'example2',\n                'a': 'parzen',\n                'b': -21\n            },\n            'messages': ['Message 5', 'Message 6'],\n            'record_type': 'process'\n        }\n    ]\n}",
   "type": "object",
   "properties": {
      "records": {
         "title": "Records",
         "default": [],
         "type": "array",
         "items": {
            "$ref": "#/definitions/Record"
         }
      }
   },
   "definitions": {
      "Record": {
         "title": "Record",
         "description": "Class to hold a record\n\nA record holds information about a process that was run. It is intended to\ntrack processes applied to data, allowing a process history to be saved\nalong with any datasets.\n\nExamples\n--------\nA simple example of creating a process record\n\n>>> from resistics.common import Record\n>>> messages = [\"message 1\", \"message 2\"]\n>>> record = Record(\n...     creator={\"name\": \"example\", \"parameter1\": 15},\n...     messages=messages,\n...     record_type=\"example\"\n... )\n>>> record.summary()\n{\n    'time_local': '...',\n    'time_utc': '...',\n    'creator': {'name': 'example', 'parameter1': 15},\n    'messages': ['message 1', 'message 2'],\n    'record_type': 'example'\n}",
         "type": "object",
         "properties": {
            "time_local": {
               "title": "Time Local",
               "type": "string",
               "format": "date-time"
            },
            "time_utc": {
               "title": "Time Utc",
               "type": "string",
               "format": "date-time"
            },
            "creator": {
               "title": "Creator",
               "type": "object"
            },
            "messages": {
               "title": "Messages",
               "type": "array",
               "items": {
                  "type": "string"
               }
            },
            "record_type": {
               "title": "Record Type",
               "type": "string"
            }
         },
         "required": [
            "creator",
            "messages",
            "record_type"
         ]
      }
   }
}

field records: List[resistics.common.Record] = []
add_record(record: resistics.common.Record)[source]

Add a process record to the list

Parameters

record (Record) – The record to add

resistics.common.get_record(creator: Dict[str, Any], messages: Union[str, List[str]], record_type: str = 'process', time_utc: Optional[datetime.datetime] = None, time_local: Optional[datetime.datetime] = None) → resistics.common.Record[source]

Get a process record

Parameters

  • creator (Dict[str, Any]) – The creator and its parameters as a dictionary

  • messages (Union[str, List[str]]) – The messages as either a single str or a list of strings

  • record_type (str, optional) – The type of record, by default “process”

  • time_utc (Optional[datetime], optional) – UTC time to attach to the record, by default None. If None, will default to UTC now

  • time_local (Optional[datetime], optional) – Local time to attach to the record, by default None. If None, will defult to local now

Returns

The process record

Return type

Record

Examples

>>> from resistics.common import get_record
>>> record = get_record(
...     creator={"name": "example", "a": 5, "b": -7.0},
...     messages="a message"
... )
>>> record.creator
{'name': 'example', 'a': 5, 'b': -7.0}
>>> record.messages
['a message']
>>> record.record_type
'process'
>>> record.time_utc
datetime.datetime(...)
>>> record.time_local
datetime.datetime(...)
resistics.common.get_history(record: resistics.common.Record, history: Optional[resistics.common.History] = None) → resistics.common.History[source]

Get a new History instance or add a record to a copy of an existing one

This method always makes a deepcopy of an input history to avoid any unplanned modifications to the inputs.

Parameters

  • record (Record) – The record

  • history (Optional[History], optional) – A history to add to, by default None

Returns

History with the record added

Return type

History

Examples

Get a new History with a single Record

>>> from resistics.common import get_history
>>> from resistics.testing import record_example1, record_example2
>>> record1 = record_example1()
>>> history = get_history(record1)
>>> history.summary()
{
    'records': [
        {
            'time_local': '...',
            'time_utc': '...',
            'creator': {
                'name': 'example1',
                'a': 5,
                'b': -7.0
            },
            'messages': ['Message 1', 'Message 2'],
            'record_type': 'process'
        }
    ]
}

Alternatively, add to an existing History. This will make a copy of the original history. If a copy is not needed, the add_record method of history can be used.

>>> record2 = record_example2()
>>> history = get_history(record2, history)
>>> history.summary()
{
    'records': [
        {
            'time_local': '...',
            'time_utc': '...',
            'creator': {
                'name': 'example1',
                'a': 5,
                'b': -7.0
            },
            'messages': ['Message 1', 'Message 2'],
            'record_type': 'process'
        },
        {
            'time_local': '...',
            'time_utc': '...',
            'creator': {
                'name': 'example2',
                'a': 'parzen',
                'b': -21
            },
            'messages': ['Message 5', 'Message 6'],
            'record_type': 'process'
        }
    ]
}
pydantic model resistics.common.ResisticsProcess[source]

Bases: resistics.common.ResisticsModel

Base class for resistics processes

Resistics processes perform operations on data (including read and write operations). Each time a ResisticsProcess child class is run, it should add a process record to the dataset

Show JSON schema
{
   "title": "ResisticsProcess",
   "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"
      }
   }
}

field name: Optional[str] [Required]
Validated by

  • validate_name

classmethod validate(value: Union[resistics.common.ResisticsProcess, Dict[str, Any]]) → resistics.common.ResisticsProcess[source]

Validate a ResisticsProcess in another pydantic class

Parameters

value (Union[ResisticsProcess, Dict[str, Any]]) – A ResisticsProcess child class or a dictionary

Returns

A ResisticsProcess child class

Return type

ResisticsProcess

Raises

  • ValueError – If the value is neither a ResisticsProcess or a dictionary

  • KeyError – If name is not in the dictionary

  • ValueError – If initialising from dictionary fails

Examples

The following example will show how a generic ResisticsProcess child class can be instantiated from ResisticsProcess using a dictionary, which might be read in from a JSON configuration file.

>>> from resistics.common import ResisticsProcess
>>> from resistics.decimate import DecimationSetup
>>> process = {"name": 'DecimationSetup', "n_levels": 8, "per_level": 5, "min_samples": 256, "div_factor": 2, "eval_freqs": None}
>>> ResisticsProcess(**process)
ResisticsProcess(name='DecimationSetup')

This is not what was expected. To get the right result, the class validate method needs to be used. This is done automatically by pydantic.

>>> ResisticsProcess.validate(process)
DecimationSetup(name='DecimationSetup', n_levels=8, per_level=5, min_samples=256, div_factor=2, eval_freqs=None)

That’s better. Note that errors will be raised if the dictionary is not formatted as expected.

>>> process = {"n_levels": 8, "per_level": 5, "min_samples": 256, "div_factor": 2, "eval_freqs": None}
>>> ResisticsProcess.validate(process)
Traceback (most recent call last):
...
KeyError: 'No name provided for initialisation of process'

This functionality is most useful in the resistics configurations which can be saved as JSON files. The default configuration uses the default parameterisation of DecimationSetup.

>>> from resistics.letsgo import Configuration
>>> config = Configuration(name="example1")
>>> config.dec_setup
DecimationSetup(name='DecimationSetup', n_levels=8, per_level=5, min_samples=256, div_factor=2, eval_freqs=None)

Now create another configuration with a different setup by passing a dictionary. In practise, this dictionary will most likely be read in from a configuration file.

>>> setup = DecimationSetup(n_levels=4, per_level=3)
>>> test_dict = setup.dict()
>>> test_dict
{'name': 'DecimationSetup', 'n_levels': 4, 'per_level': 3, 'min_samples': 256, 'div_factor': 2, 'eval_freqs': None}
>>> config2 = Configuration(name="example2", dec_setup=test_dict)
>>> config2.dec_setup
DecimationSetup(name='DecimationSetup', n_levels=4, per_level=3, min_samples=256, div_factor=2, eval_freqs=None)

This method allows the saving of a configuration with custom processors in a JSON file which can be loaded and used again.

parameters() → Dict[str, Any][source]

Return any process parameters incuding the process name

These parameters are expected to be primatives and should be sufficient to reinitialise the process and re-run the data. The base class assumes all class variables meet this description.

Returns

Dictionary of parameters

Return type

Dict[str, Any]

class resistics.common.ResisticsBase[source]

Bases: object

Resistics base class

Parent class to ensure consistency of common methods

type_to_string() → str[source]

Get the class type as a string

to_string() → str[source]

Class details as a string

summary(symbol: str = '-') → None[source]

Print a summary of class details

class resistics.common.ResisticsData[source]

Bases: resistics.common.ResisticsBase

Base class for a resistics data object

pydantic model resistics.common.ResisticsWriter[source]

Bases: resistics.common.ResisticsProcess

Parent process for data writers

Parameters

overwrite (bool, optional) – Boolean flag for overwriting the existing data, by default False

Show JSON schema
{
   "title": "ResisticsWriter",
   "description": "Parent process for data writers\n\nParameters\n----------\noverwrite : bool, optional\n    Boolean flag for overwriting the existing data, by default False",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "overwrite": {
         "title": "Overwrite",
         "default": true,
         "type": "boolean"
      }
   }
}

field overwrite: bool = True
run(dir_path: pathlib.Path, data: resistics.common.ResisticsData) → None[source]

Write out a ResisticsData child object to a directory