Source code for scatcluster.structure
# scatcluster/structure.py
import json
import os
from glob import glob
from typing import List, Optional
[docs]class Structure:
[docs] def prepare_directory_structure(self, directories: Optional[List[str]] = None) -> None:
"""Build directory structure required for workflow processing
Args:
directories (List[str], optional): Directories created as part of the workflow.
Defaults to ['scatterings', 'figures', 'networks', 'ICA','clustering','data', 'config'].
"""
if directories is None:
directories = ['scatterings', 'figures', 'networks', 'ICA', 'clustering', 'data', 'config']
if self.data_savepath == '':
raise ValueError('"data_savepath" has not been set up correctly. Kindly update the parametrization.')
if os.path.exists(self.data_savepath):
print(f'Main directory {self.data_savepath} already exists. \n')
else:
os.mkdir(self.data_savepath)
print(f'Main directory {self.data_savepath} created. \n')
for path in directories:
isExist = os.path.exists(f'{self.data_savepath}{path}')
if not isExist:
os.makedirs(f'{self.data_savepath}{path}')
print(f'Directory {self.data_savepath}{path} created.')
else:
print(f'Directory {self.data_savepath}{path} already exists.')
[docs] def delete_scatterings(self) -> None:
"""
Deletes scatterings files based on specific naming conventions.
"""
file_list_scatterings = glob(
f'{self.data_savepath}scatterings/{self.data_network}_{self.data_station}_{self.data_location}_'
f'{self.network_name}_scatterings_*.npz')
for file in file_list_scatterings:
os.remove(file)
print(f'{len(file_list_scatterings)} scatterings have been deleted from "{self.data_savepath}scatterings/"\n')
[docs] def config_store(self):
"""
Stores the configuration of the current object as a JSON file.
This function creates a dictionary `sc_config` containing the configuration parameters of the current object.
The dictionary includes the following keys:
- 'data_savepath': The path to the main directory where data is saved.
- 'data_client_path': The path to the client data.
- 'data_network': The network name.
- 'data_station': The station name.
- 'data_location': The location name.
- 'data_channel': The channel name.
- 'data_sample_starttime': The start time of the data sample.
- 'data_sample_endtime': The end time of the data sample.
- 'data_starttime': The start time of the data.
- 'data_endtime': The end time of the data.
- 'data_exclude_days': The excluded days for the data.
- 'network_segment': The network segment.
- 'network_step': The network step.
- 'network_sampling_rate': The network sampling rate.
- 'network_banks': The network banks.
- 'network_pooling': The network pooling.
- 'ica_ev_limit': The ICA event limit.
- 'ica_min_ICAs': The minimum number of ICAs.
- 'ica_max_ICAs': The maximum number of ICAs.
- 'ica_overwrite_previous_models': Whether to overwrite previous models.
- 'dendrogram_method': The method for generating the dendrogram.
- 'waveforms_n_samples': The number of samples for waveforms.
The function then constructs the path to the JSON file by combining the values of 'data_savepath',
'data_network', 'data_station', 'data_location', and 'network_name'. The JSON file is opened in write mode with
UTF-8 encoding. The `sc_config` dictionary is dumped into the JSON file using the `json.dump()` function.
Finally, a message is printed indicating the location where the JSON file was stored.
Parameters:
- self: The current object.
Returns:
- None
"""
sc_config = {
'data_savepath': self.data_savepath,
'data_client_path': self.data_client_path,
'data_network': self.data_network,
'data_station': self.data_station,
'data_location': self.data_location,
'data_channel': self.data_channel,
'data_sample_starttime': self.data_sample_starttime,
'data_sample_endtime': self.data_sample_endtime,
'data_starttime': self.data_starttime,
'data_endtime': self.data_endtime,
'data_exclude_days': self.data_exclude_days,
'network_segment': self.network_segment,
'network_step': self.network_step,
'network_sampling_rate': self.network_sampling_rate,
'network_banks': self.network_banks,
'network_pooling': self.network_pooling,
'ica_ev_limit': self.ica_ev_limit,
'ica_min_ICAs': self.ica_min_ICAs,
'ica_max_ICAs': self.ica_max_ICAs,
'ica_overwrite_previous_models': self.ica_overwrite_previous_models,
'dendrogram_method': self.dendrogram_method,
'waveforms_n_samples': self.waveforms_n_samples
}
ssn_json_file = (f'{self.data_savepath}config/{self.data_network}_{self.data_station}_{self.data_location}_'
f'{self.network_name}.json')
with open(ssn_json_file, 'w', encoding='utf8') as jsonfile:
json.dump(sc_config, jsonfile)
print(f'SSN config stored at "{ssn_json_file}"')
[docs] def config_show(self):
"""
Prints the attributes of the object as a dictionary.
This function prints the attributes of the object as a dictionary using the `__dict__` attribute. The attributes
are the instance variables of the object.
Parameters:
self (object): The object whose attributes are to be printed.
"""
print(self.__dict__)
[docs] def list_available_configurations(self):
"""
Lists all the available configurations in the config directory.
This function uses the `glob` function to find all the JSON files in the config directory.
It prints the list of JSON files found.
Parameters:
self (object): The instance of the class.
"""
print(glob(f'{self.data_savepath}config/*json'))