resistics.common module#
Common resistics functions and classes used throughout the package
- resistics.common.is_file(file_path: 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:
- resistics.common.assert_file(file_path: 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: 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:
- resistics.common.assert_dir(dir_path: 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: Path) Tuple[List[Path], List[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:
PathNotFoundError – Path does not exist
NotDirectoryError – Path is not a directory
- resistics.common.dir_files(dir_path: Path) List[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:
- resistics.common.dir_subdirs(dir_path: Path) List[Path] [source]#
Get subdirectories in directory
Excludes hidden files
- Parameters:
dir_path (Path) – Parent directory path
- Returns:
dirs – List of subdirectories
- Return type:
- resistics.common.known_chan(chan: str) bool [source]#
Check whether resistics is familiar with a channel name
- Parameters:
chan (str) – The channel name
- Returns:
True if it is a resistics known channel, false otherwise
- Return type:
Examples
>>> from resistics.common import known_chan >>> known_chan("Ex") True >>> known_chan("Hy") True >>> known_chan("cat") False
- resistics.common.is_electric(chan: str) bool [source]#
Check if a channel is electric
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
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:
- 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:
- Returns:
True if all checks passed
- Return type:
- 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:
Examples
>>> from resistics.common import fs_to_string >>> fs_to_string(512.0) '512_000000'
- resistics.common.array_to_string(data: ndarray, sep: str = ', ', precision: int = 8, scientific: bool = False) str [source]#
Convert an array to a string for logging or printing
- Parameters:
- Returns:
String representation of array
- Return type:
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:
BaseModel
Base resistics model
Show JSON schema
{ "title": "ResisticsModel", "description": "Base resistics model", "type": "object", "properties": {} }
- pydantic model resistics.common.ResisticsFile[source]#
Bases:
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" } } }
- pydantic model resistics.common.Metadata[source]#
Bases:
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:
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: ResisticsFile | None = None#
Information about a file, relevant if writing out or reading back in
- pydantic model resistics.common.Record[source]#
Bases:
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" ] }
- pydantic model resistics.common.History[source]#
Bases:
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" ] } } }
- resistics.common.get_record(creator: Dict[str, Any], messages: str | List[str], record_type: str = 'process', time_utc: datetime | None = None, time_local: datetime | None = None) 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:
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: Record, history: History | None = None) 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:
- Returns:
History with the record added
- Return type:
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:
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" } } }
- classmethod validate(value: ResisticsProcess | Dict[str, Any]) 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:
- 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
- class resistics.common.ResisticsData[source]#
Bases:
ResisticsBase
Base class for a resistics data object
- pydantic model resistics.common.ResisticsWriter[source]#
Bases:
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" } } }
- run(dir_path: Path, data: ResisticsData) None [source]#
Write out a ResisticsData child object to a directory