1 clim_recal.utils.core
clim_recal.utils.core
Utility functions.
1.1 Classes
| Name | Description |
|---|---|
| MonthDay | A class to ease generating annual time series. |
1.1.1 MonthDay
clim_recal.utils.core.MonthDay(self, month=1, day=1, format_str=ISO_DATE_FORMAT_STR)
A class to ease generating annual time series.
1.1.1.1 Attributes
| Name | Type | Description |
|---|---|---|
| month | int | What Month as an integer from 1 to 12. |
| day | int | What day in the month, an integer from 1 to 31. |
| format_str | str | Format to use if generating a str. |
1.1.1.2 Examples
>>> jan_1: MonthDay = MonthDay()
>>> jan_1.from_year(1980)
datetime.date(1980, 1, 1)
>>> jan_1.from_year('1980')
datetime.date(1980, 1, 1)
>>> jan_1.from_year('1980', as_str=True)
'1980-01-01'1.1.1.3 Methods
| Name | Description |
|---|---|
| from_year | Return a date or str of date given year. |
| from_year_range_to_str | Return an annual range str. |
1.1.1.3.1 from_year
clim_recal.utils.core.MonthDay.from_year(year, as_str=False)
Return a date or str of date given year.
1.1.1.3.2 from_year_range_to_str
clim_recal.utils.core.MonthDay.from_year_range_to_str(start_year, end_year, split_str=DATE_FORMAT_SPLIT_STR, in_format_str=CLI_DATE_FORMAT_STR, out_format_str=CLI_DATE_FORMAT_STR, include_end_date=True)
Return an annual range str.
1.1.1.3.2.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
start_year |
int | str | Starting year to combine with self.month and self.day. |
required |
end_year |
int | str | Starting year to combine with self.month and self.day. |
required |
split_str |
str | str between start_date and end_date generated. |
DATE_FORMAT_SPLIT_STR |
include_end_date |
bool | Whether to include the end_date in the final str. If False follow python convention to return the day prior. |
True |
1.1.1.3.2.2 Examples
>>> jan_1: MonthDay = MonthDay()
>>> jan_1.from_year_range_to_str(1980, 1982, include_end_date=False)
'19800101-19811231'
>>> jan_1.from_year_range_to_str('1980', 2020)
'19800101-20200101'1.2 Functions
| Name | Description |
|---|---|
| annual_data_path | Return Path of annual data files. |
| annual_data_paths_generator | Yield Path of annual data files. |
| check_package_path | Return path for test running. |
| climate_data_mount_path | Return likely climate data mount path based on operating system. |
| csv_reader | Yield a dict per row from a CSV file at path. |
| date_range_generator | Return a tuple of DateType objects. |
| date_range_to_str | Take start_date and end_date and return a date range str. |
| date_to_str | Return a str in date_format_str of date_obj. |
| ensure_date | Ensure passed date_to_check is a date. |
| is_climate_data_mounted | Check if CLIMATE_DATA_MOUNT_PATH is mounted. |
| is_platform_darwin | Check if sys.platform is Darwin (macOS). |
| iter_to_tuple_strs | Return a tuple with all components converted to strs. |
| multiprocess_execute | Run method_name as from iter via multiprocessing. |
| path_iterdir | Return an Generator after ensuring path exists. |
| product_dict | Return product combinatorics of kwargs. |
| results_path | Return Path: path/name_time.extension. |
| run_callable_attr | Extract method_name from instance to call. |
| time_str | Return a str of passed or generated time suitable for a file name. |
1.2.1 annual_data_path
clim_recal.utils.core.annual_data_path(start_year=1980, end_year=1981, month_day=DEFAULT_CPM_START_MONTH_DAY, include_end_date=False, parent_path=None, file_name_middle_str=CPM_FILE_NAME_MIDDLE_STR, file_name_extension=NETCDF_EXTENSION, data_type=MAX_TEMP_STR, make_paths=False)
Return Path of annual data files.
1.2.1.1 Examples
>>> str(annual_data_path(parent_path=Path('test/path')))
'test/path/tasmax_rcp85_land-cpm_uk_2.2km_01_day_19801201-19811130.nc'1.2.2 annual_data_paths_generator
clim_recal.utils.core.annual_data_paths_generator(start_year=1980, end_year=1986, **kwargs)
Yield Path of annual data files.
1.2.2.1 Examples
>>> pprint(tuple(annual_data_paths_generator()))
(PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19801201-19811130.nc'),
PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19811201-19821130.nc'),
PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19821201-19831130.nc'),
PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19831201-19841130.nc'),
PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19841201-19851130.nc'),
PosixPath('tasmax_rcp85_land-cpm_uk_2.2km_01_day_19851201-19861130.nc'))
>>> parent_path_example: Iterator[Path] = annual_data_paths_generator(
... parent_path=Path('test/path'))
>>> str(next(parent_path_example))
'test/path/tasmax_rcp85_land-cpm_uk_2.2km_01_day_19801201-19811130.nc'1.2.3 check_package_path
clim_recal.utils.core.check_package_path(strict=True, try_chdir=False)
Return path for test running.
1.2.3.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
strict |
bool | Whether to raise a ValueError if check fails. |
True |
try_chdir |
bool | Whether to attempt changing directory if initial check fails | False |
1.2.3.2 Raises
| Type | Description |
|---|---|
| ValueError | If strict and checks fail. |
1.2.3.3 Returns
| Type | Description |
|---|---|
| Whether to check if call was successful. |
1.2.3.4 Examples
>>> check_package_path()
True
>>> chdir('..')
>>> check_package_path(strict=False)
False
>>> check_package_path()
Traceback (most recent call last):
...
ValueError: 'clim-recal' pipeline must be run in...
>>> check_package_path(try_chdir=True)
True1.2.4 climate_data_mount_path
clim_recal.utils.core.climate_data_mount_path(is_darwin=None, full_path=True)
Return likely climate data mount path based on operating system.
1.2.4.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
is_darwin |
bool | None | Whether to use CLIMATE_DATA_MOUNT_PATH_DARWIN or call is_platform_darwin if None. fixture is_platform_darwin. |
None |
1.2.4.2 Returns
| Type | Description |
|---|---|
The Path climate data would likely be mounted to. |
1.2.5 csv_reader
clim_recal.utils.core.csv_reader(path, **kwargs)
Yield a dict per row from a CSV file at path.
1.2.5.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
path |
PathLike | CSV file Path. |
required |
**kwargs |
Additional parameters for csv.DictReader. |
{} |
1.2.5.2 Examples
>>> import csv
>>> csv_path: Path = TEST_AUTH_CSV_PATH
>>> auth_dict: dict[str, str] = {
... 'sally': 'fig*new£kid',
... 'george': 'tee&iguana*sky',
... 'susan': 'history!bill-walk',}
>>> field_names: tuple[str, str] = ('user_name', 'password')
>>> with open(csv_path, 'w') as csv_file:
... writer = csv.writer(csv_file)
... line_num: int = writer.writerow(('user_name', 'password'))
... for user_name, password in auth_dict.items():
... line_num = writer.writerow((user_name, password))
>>> tuple(csv_reader(csv_path))
({'user_name': 'sally', 'password': 'fig*new£kid'},
{'user_name': 'george', 'password': 'tee&iguana*sky'},
{'user_name': 'susan', 'password': 'history!bill-walk'})1.2.6 date_range_generator
clim_recal.utils.core.date_range_generator(start_date, end_date, inclusive=False, skip_dates=None, start_format_str=CLI_DATE_FORMAT_STR, end_format_str=CLI_DATE_FORMAT_STR, output_format_str=CLI_DATE_FORMAT_STR, skip_dates_format_str=CLI_DATE_FORMAT_STR, yield_type=date)
Return a tuple of DateType objects.
1.2.6.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
start_date |
DateType | DateType at start of time series. |
required |
end_date |
DateType | DateType at end of time series. |
required |
inclusive |
bool | Whether to include the end_date in the returned time series. |
False |
skip_dates |
Iterable[DateType] | DateType | None | Dates to skip between start_date and end_date. |
None |
start_format_str |
str | A strftime format to apply if start_date type is str. |
CLI_DATE_FORMAT_STR |
end_format_str |
str | A strftime format to apply if end_date type is str. |
CLI_DATE_FORMAT_STR |
output_format_str |
str | A strftime format to apply if yield_type is str. |
CLI_DATE_FORMAT_STR |
skip_dates_format_str |
str | A strftime format to apply if any skip_dates are str. |
CLI_DATE_FORMAT_STR |
yield_type |
type[date] | type[str] | Whether which date type to return in tuple (date or str). |
date |
1.2.6.2 Returns
| Type | Description |
|---|---|
| Iterator[DateType] | A tuple of date or str objects (only one type throughout). |
1.2.6.3 Examples
>>> four_years: tuple[date] = tuple(date_range_generator('19801130', '19841130'))
>>> len(four_years)
1461
>>> four_years_inclusive: tuple[date] = tuple(
... date_range_generator('1980-11-30', '19841130',
... inclusive=True,
... start_format_str=ISO_DATE_FORMAT_STR))
>>> len(four_years_inclusive)
1462
>>> four_years_inclusive_skip: tuple[date] = tuple(
... date_range_generator('19801130', '19841130',
... inclusive=True,
... skip_dates='19840229'))
>>> len(four_years_inclusive_skip)
1461
>>> skip_dates: tuple[date] = (date(1981, 12, 1), date(1982, 12, 1))
>>> four_years_inclusive_skip: tuple[date] = list(
... date_range_generator('19801130', '19841130',
... inclusive=True,
... skip_dates=skip_dates))
>>> len(four_years_inclusive_skip)
1460
>>> skip_dates in four_years_inclusive_skip
False1.2.7 date_range_to_str
clim_recal.utils.core.date_range_to_str(start_date, end_date, split_str=DATE_FORMAT_SPLIT_STR, in_format_str=CLI_DATE_FORMAT_STR, out_format_str=CLI_DATE_FORMAT_STR, include_end_date=True)
Take start_date and end_date and return a date range str.
1.2.7.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
start_date |
DateType | First date in range. | required |
end_date |
DateType | Last date in range | required |
split_str |
str | char to split returned date range str. |
DATE_FORMAT_SPLIT_STR |
in_format_str |
str | A strftime format str to convert start_date from. |
CLI_DATE_FORMAT_STR |
out_format_str |
str | A strftime format str to convert end_date from. |
CLI_DATE_FORMAT_STR |
1.2.7.2 Returns
| Type | Description |
|---|---|
| str | A str of date range from start_date to end_date in the out_format_str format. |
1.2.7.3 Examples
>>> date_range_to_str('20100101', '20100330')
'20100101-20100330'
>>> date_range_to_str(date(2010, 1, 1), '20100330')
'20100101-20100330'
>>> date_range_to_str(date(2010, 1, 1), '20100330', include_end_date=False)
'20100101-20100329'1.2.8 date_to_str
clim_recal.utils.core.date_to_str(date_obj, in_format_str=CLI_DATE_FORMAT_STR, out_format_str=CLI_DATE_FORMAT_STR)
Return a str in date_format_str of date_obj.
1.2.8.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
date_obj |
DateType | A datetime.date or str object to convert. |
required |
in_format_str |
str | A strftime format str to convert date_obj from if date_obj is a str. |
CLI_DATE_FORMAT_STR |
out_format_str |
str | A strftime format str to convert date_obj to. |
CLI_DATE_FORMAT_STR |
1.2.8.2 Returns
| Type | Description |
|---|---|
| str | A str version of date_obj in out_format_str format. |
1.2.8.3 Examples
>>> date_to_str('20100101')
'20100101'
>>> date_to_str(date(2010, 1, 1))
'20100101'1.2.9 ensure_date
clim_recal.utils.core.ensure_date(date_to_check, format_str=CLI_DATE_FORMAT_STR)
Ensure passed date_to_check is a date.
1.2.9.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
date_to_check |
DateType | Date or str to ensure is a date. |
required |
format_str |
str | strptime str to convert date_to_check if necessary. |
CLI_DATE_FORMAT_STR |
1.2.9.2 Returns
| Type | Description |
|---|---|
date instance from date_to_check. |
1.2.9.3 Examples
>>> ensure_date('19801130')
datetime.date(1980, 11, 30)
>>> ensure_date(date(1980, 11, 30))
datetime.date(1980, 11, 30)1.2.10 is_climate_data_mounted
clim_recal.utils.core.is_climate_data_mounted(mount_path=None)
Check if CLIMATE_DATA_MOUNT_PATH is mounted.
1.2.10.1 Notes
Consider refactoring to climate_data_mount_path returns None when conditions here return False.
1.2.11 is_platform_darwin
clim_recal.utils.core.is_platform_darwin()
Check if sys.platform is Darwin (macOS).
1.2.12 iter_to_tuple_strs
clim_recal.utils.core.iter_to_tuple_strs(iter_var, func=str)
Return a tuple with all components converted to strs.
1.2.12.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
iter_var |
Iterable[Any] | Iterable of objects that can be converted into strs. |
required |
func |
Callable[[Any], str] | Callable to convert each obj in iter_val to a str. |
str |
1.2.12.2 Returns
| Type | Description |
|---|---|
| tuple[str, …] | A tuple of all iter_var elements in str format. |
1.2.12.3 Examples
>>> iter_to_tuple_strs(['cat', 1, Path('a/path')])
('cat', '1', 'a/path')
>>> iter_to_tuple_strs(
... ['cat', 1, Path('a/path')],
... lambda x: f'{x:02}' if type(x) == int else str(x))
('cat', '01', 'a/path')1.2.13 multiprocess_execute
clim_recal.utils.core.multiprocess_execute(iter, *args, method_name=DEFAULT_CALLABLE_ATTR_NAME, progress_bar=False, bar_name='', cpus=None, include_sub_process_config=False, sub_process_progress_bar=False, **kwargs)
Run method_name as from iter via multiprocessing.
1.2.13.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
iter |
Sequence | Sequence of instances to iterate over calling method_name from. |
required |
*args |
Any | Args to pass to method_name |
() |
method_name |
str | What to call from objects in inter. |
DEFAULT_CALLABLE_ATTR_NAME |
progress_bar |
bool | Whether to render a progress bar. | False |
bar_name |
str | Passed to description parameter for progress bar. |
'' |
cpus |
int | None | Number of cpus to pass to Pool for multiprocessing. |
None |
include_sub_process_config |
bool | Include configurations specific to multiprocessess (eg. progress_bar config) |
False |
sub_process_progress_bar |
bool | Whether to include progress_bar within each multiprocess run. |
False |
**kwargs |
Any | Args to pass to method_name |
{} |
1.2.13.2 Examples
>>> test_strs: tuple[tuple[str], ...] = (('the third'), ('man'))
>>> multiprocess_execute(test_strs, method_name="split")
[['the', 'third'], ['man']]
>>> test_strs = (('lumbar'), ('puncture'))
>>> multiprocess_execute(test_strs, 'u', method_name="split")
[['l', 'mbar'], ['p', 'nct', 're']]
>>> multiprocess_execute(test_strs, 'u', method_name="split",
... progress_bar=True)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%...
[['l', 'mbar'], ['p', 'nct', 're']]1.2.13.3 Notes
Failed asserting cpus <= total - 1.
1.2.14 path_iterdir
clim_recal.utils.core.path_iterdir(path, strict=False)
Return an Generator after ensuring path exists.
1.2.14.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
path |
PathLike | Path to folder to iterate through. |
required |
strict |
bool | Whether to raise FileNotFoundError if path not found. |
False |
1.2.14.2 Raises
| Type | Description |
|---|---|
| FileNotFoundError | Raised if strict = True and path does not exist. |
1.2.14.3 Returns
| Type | Description |
|---|---|
| Generator[Optional[Path], None, None] | None if FileNotFoundError error and strict is False. |
1.2.14.4 Examples
>>> tmp_path = getfixture('tmp_path')
>>> from os import chdir
>>> chdir(tmp_path)
>>> example_path: Path = Path('a/test/path')
>>> example_path.exists()
False
>>> tuple(path_iterdir(example_path.parent))
()
>>> tuple(path_iterdir(example_path.parent, strict=True))
Traceback (most recent call last):
...
FileNotFoundError: [Errno 2] No such file or directory: 'a/test'
>>> example_path.parent.mkdir(parents=True)
>>> example_path.touch()
>>> tuple(path_iterdir(example_path.parent))
(PosixPath('a/test/path'),)
>>> example_path.unlink()
>>> tuple(path_iterdir(example_path.parent))
()1.2.15 product_dict
clim_recal.utils.core.product_dict(**kwargs)
Return product combinatorics of kwargs.
1.2.15.1 Examples
>>> pprint(tuple(
... product_dict(animal=['cat', 'fish'], activity=('swim', 'eat'))))
({'activity': 'swim', 'animal': 'cat'},
{'activity': 'eat', 'animal': 'cat'},
{'activity': 'swim', 'animal': 'fish'},
{'activity': 'eat', 'animal': 'fish'})1.2.16 results_path
clim_recal.utils.core.results_path(name, path=None, time=None, extension=None, mkdir=False, dot_pre_extension=True)
Return Path: path/name_time.extension.
1.2.16.1 Examples
>>> str(results_path('hads', 'test_example/folder', extension='cat'))
'test_example/folder/hads_..._...-....cat'1.2.17 run_callable_attr
clim_recal.utils.core.run_callable_attr(instance, *args, method_name=DEFAULT_CALLABLE_ATTR_NAME, **kwargs)
Extract method_name from instance to call.
1.2.17.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
instance |
object | object to call method_name from. |
required |
*args |
Parameters passed to method_name from instance. |
() |
|
method_name |
str | Name of method on object to call. |
DEFAULT_CALLABLE_ATTR_NAME |
**kwargs |
Parameters passed to method_name from instance. |
{} |
1.2.17.2 Notes
This is primarily meant to address issues with pickle, particularly when using multiprocessing and lambda functions yield pickle errors.
1.2.17.3 Examples
>>> jan_1: MonthDay = MonthDay()
>>> run_callable_attr(jan_1, 1984, method_name='from_year')
datetime.date(1984, 1, 1)1.2.18 time_str
clim_recal.utils.core.time_str(time=None, format=RUN_TIME_STAMP_FORMAT, replace_char_tuple=None)
Return a str of passed or generated time suitable for a file name.
1.2.18.1 Parameters
| Name | Type | Description | Default |
|---|---|---|---|
time |
date | datetime | None | Time to format. Will be set to current time if None is passed, and add current time if a date is passed to convert to a datetime. |
None |
format |
str | strftime str format to return time as. If replace_chars is passed, these will be used to replace those in strftime. |
RUN_TIME_STAMP_FORMAT |
replace_char_tuple |
tuple[str, str] | None | tuple of (char_to_match, char_to_replace) order. |
None |
1.2.18.2 Examples
>>> time_str(datetime(2024, 10, 10, 10, 10, 10))
'24-10-10_10-10'