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.
range_len	Cacluate the total length of range with indexing.
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)
True

1.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
False

1.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, method_name=None, cpus=None)

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
method_name	str | None	What to call from objects in inter.	None
cpus	int | None	Number of cpus to pass to Pool for multiprocessing.	None

1.2.13.2 Examples

>>> if not is_data_mounted:
...     pytest.skip(mount_doctest_skip_message)
>>> from clim_recal.resample import CPMResampler
>>> resample_test_hads_output_path: Path = getfixture(
...         'resample_test_cpm_output_path')
>>> cpm_resampler: CPMResampler = CPMResampler(
...     stop_index=3,
...     output_path=resample_test_hads_output_path,
... )
>>> multiprocess_execute(cpm_resampler, method_name="exists")
[True, True, True]

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 range_len

clim_recal.utils.core.range_len(maximum, start=0, stop=None, step=1)

Cacluate the total length of range with indexing.

1.2.16.1 Parameters

Name	Type	Description	Default
maximum	int	Maximum range length.	required
start	int	Index to start from.	0
stop	int | None	Index to stop at.	None
step	int	Steps between start and stop indexes	1

1.2.16.2 Examples

>>> range_len(100)
100
>>> range_len(100, 90)
10
>>> range_len(100, 20, 30)
10
>>> range_len(100, 20, 30, 2)
5

1.2.17 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.17.1 Examples

>>> str(results_path('hads', 'test_example/folder', extension='cat'))
'test_example/folder/hads_..._...-....cat'

1.2.18 run_callable_attr

clim_recal.utils.core.run_callable_attr(instance, method_name='execute', *args, **kwargs)

Extract method_name from instance to call.

1.2.18.1 Parameters

Name	Type	Description	Default
instance	object	object to call method_name from.	required
method_name	str	Name of method on object to call.	'execute'
*args		Parameters passed to method_name from instance.	()
**kwargs		Parameters passed to method_name from instance.	{}

1.2.18.2 Notes

This is primarily meant to address issues with pickle, particularly when using multiprocessing and lambda functions yield pickle errors.

1.2.18.3 Examples

>>> jan_1: MonthDay = MonthDay()
>>> run_callable_attr(jan_1, 'from_year', 1984)
datetime.date(1984, 1, 1)

1.2.19 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.19.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.19.2 Examples

>>> time_str(datetime(2024, 10, 10, 10, 10, 10))
'24-10-10_10-10'