diff --git a/python/idsse_common/idsse/common/log_util.py b/python/idsse_common/idsse/common/log_util.py index fbaf6f84..7f7ded2b 100644 --- a/python/idsse_common/idsse/common/log_util.py +++ b/python/idsse_common/idsse/common/log_util.py @@ -28,7 +28,7 @@ def set_corr_id_context_var( originator: str, - key: uuid.UUID = uuid.uuid4(), + key: uuid.UUID = uuid.UUID('00000000-0000-0000-0000-000000000000'), issue_dt: str | datetime | None = None ) -> None: """ diff --git a/python/idsse_common/idsse/common/path_builder.py b/python/idsse_common/idsse/common/path_builder.py index 92a52861..2c77cd76 100644 --- a/python/idsse_common/idsse/common/path_builder.py +++ b/python/idsse_common/idsse/common/path_builder.py @@ -18,30 +18,49 @@ import os import re +from copy import deepcopy from datetime import datetime, timedelta, UTC -from typing import Self +from typing import Final, NamedTuple, Self from .utils import TimeDelta +# The public class class PathBuilder: """Path Builder Class""" + ISSUE: Final[str] = 'issue' + VALID: Final[str] = 'valid' + LEAD: Final[str] = 'lead' + INT: Final[str] = 'd' + FLOAT: Final[str] = 'f' + STR: Final[str] = 's' + def __init__(self, basedir: str, subdir: str, file_base: str, file_ext: str) -> None: - self._basedir = basedir - self._subdir = subdir + + # store path format parts to private vars, they accessible via properties + self._base_dir = basedir + self._sub_dir = subdir self._file_base = file_base self._file_ext = file_ext + # create a dictionary to hold lookup info + self._lookup_dict = {} + self._update_lookup(self.path_fmt) + + # these are used for caching the most recent previously parsed paths (for optimization) + self._last_parsed_path = None + self._parsed_args = None + def __str__(self) -> str: - return f"'{self._basedir}','{self._subdir}','{self._file_base}','{self._file_ext}'" + return f"'{self._base_dir}','{self._sub_dir}','{self._file_base}','{self._file_ext}'" def __repr__(self) -> str: - return (f"PathBuilder(basedir='{self._basedir}', subdir='{self._subdir}', " + return (f"PathBuilder(basedir='{self._base_dir}', subdir='{self._sub_dir}', " f"file_base='{self._file_base}', file_ext='{self._file_ext}')") @classmethod @@ -73,51 +92,80 @@ def from_path(cls, path_fmt: str) -> Self: Self: The newly created PathBuilder object """ idx = path_fmt.rindex(os.path.sep) - return PathBuilder(path_fmt[:idx], '', path_fmt[:idx], '') + return PathBuilder(path_fmt[:idx], '', path_fmt[idx+1:], '') @property def dir_fmt(self): - """Provides access to the directory format str""" - return os.path.join(self._basedir, self._subdir) + """Provides access to the directory format string""" + return os.path.join(self.base_dir, self.sub_dir) @property def filename_fmt(self): - """Provides access to the filename format str""" - if not self._file_ext or self._file_ext.startswith('.'): - return f'{self._file_base}{self._file_ext}' - return f'{self._file_base}.{self._file_ext}' + """Provides access to the filename format string""" + if not self.file_ext or self.file_ext.startswith('.'): + return f'{self.file_base}{self.file_ext}' + return f'{self.file_base}.{self.file_ext}' + + @property + def path_fmt(self): + """Provides access to the path format string""" + return os.path.join(self.dir_fmt, self.filename_fmt) @property def base_dir(self): - """Provides access to the file base directory format str""" - return self._basedir + """Provides access to the file base directory format string""" + return self._base_dir @base_dir.setter def base_dir(self, basedir): - """Set the file extension format str""" - self._basedir = basedir + """Set the base directory format string""" + # update base directory format + self._base_dir = basedir + self._update_lookup(self.path_fmt) + + @property + def sub_dir(self): + """Provides access to the file base directory format string""" + return self._sub_dir + + @sub_dir.setter + def sub_dir(self, subdir): + """Set the sub directory format string""" + # update sub directory format + self._sub_dir = subdir + self._update_lookup(self.path_fmt) + + @property + def file_base(self): + """Provides access to the file base format string""" + return self._file_base + + @file_base.setter + def file_base(self, file_base): + """Set the file extension format string""" + # update file extension format + self._file_base = file_base + self._update_lookup(self.path_fmt) @property def file_ext(self): - """Provides access to the file extension format str""" + """Provides access to the file extension format string""" if self._file_ext: return self._file_ext return self._file_base[self._file_base.rindex('.'):] @file_ext.setter def file_ext(self, ext): - """Set the file extension format str""" + """Set the file extension format string""" + # update file extension format self._file_ext = ext - - @property - def path_fmt(self): - """Provides access to the path format str""" - return os.path.join(self.dir_fmt, self.filename_fmt) + self._update_lookup(self.path_fmt) def build_dir(self, issue: datetime | None = None, valid: datetime | None = None, - lead: timedelta | TimeDelta | None = None) -> str: + lead: timedelta | TimeDelta | None = None, + **kwargs) -> str: """Attempts to build the directory with provided arguments Args: @@ -127,6 +175,7 @@ def build_dir(self, directory is dependant on it. . Defaults to None. lead (timedelta | TimeDelta | None, optional): Lead can be provided in addition to issue or valid. Defaults to None. + **kwargs: Any additional key/word args (i.e. 'region'='co') Returns: str: Directory as a string @@ -134,12 +183,13 @@ def build_dir(self, if issue is None: return None lead = self._ensure_lead(issue, valid, lead) - return self.dir_fmt.format(issue=issue, valid=valid, lead=lead) + return self.dir_fmt.format(issue=issue, valid=valid, lead=lead, **kwargs) def build_filename(self, issue: datetime | None = None, valid: datetime | None = None, - lead: timedelta | TimeDelta | None = None) -> str: + lead: timedelta | TimeDelta | None = None, + **kwargs) -> str: """Attempts to build the filename with provided arguments Args: @@ -149,17 +199,19 @@ def build_filename(self, filename is dependant on it. . Defaults to None. lead (timedelta | TimeDelta | None, optional): Lead can be provided in addition to issue or valid. Defaults to None. + **kwargs: Any additional key/word args (i.e. 'region'='co') Returns: str: File name as a string """ lead = self._ensure_lead(issue, valid, lead) - return self.filename_fmt.format(issue=issue, valid=valid, lead=lead) + return self.filename_fmt.format(issue=issue, valid=valid, lead=lead, **kwargs) def build_path(self, issue: datetime | None = None, valid: datetime | None = None, - lead: timedelta | TimeDelta | None = None) -> str: + lead: timedelta | TimeDelta | None = None, + **kwargs: dict) -> str: """Attempts to build the path with provided arguments Args: @@ -169,26 +221,27 @@ def build_path(self, path is dependant on it. . Defaults to None. lead (timedelta | TimeDelta | None, optional): Lead can be provided in addition to issue or valid. Defaults to None. + **kwargs: Any additional key/word args (i.e. 'region'='co') Returns: str: Path as a string """ lead = self._ensure_lead(issue, valid, lead) - return self.path_fmt.format(issue=issue, valid=valid, lead=lead) + return self._apply_format(self.path_fmt, issue=issue, valid=valid, lead=lead, **kwargs) - def parse_dir(self, dir_: str) -> dict: + def parse_dir(self, dir_str: str) -> dict: """Extracts issue, valid, and/or lead information from the provided directory Args: - dir_ (str): A directory consistent with this PathBuilder + dir_str (str): A directory consistent with this PathBuilder Returns: dict: Lookup of all information identified and extracted """ - return self._parse_times(dir_, self.dir_fmt) + return self._get_parsed_arg_parts(dir_str, self.dir_fmt) def parse_filename(self, filename: str) -> dict: - """Extracts issue, valid, and/or lead information from the provided filename + """Extracts issue, valid, lead, and any extras information from the provided filename Args: filename (str): A filename consistent with this PathBuilder @@ -197,10 +250,11 @@ def parse_filename(self, filename: str) -> dict: dict: Lookup of all information identified and extracted """ filename = os.path.basename(filename) - return self._parse_times(filename, self.filename_fmt) + self._parse_path(filename, self.filename_fmt) + return deepcopy(self._parsed_args) def parse_path(self, path: str) -> dict: - """Extracts issue, valid, and/or lead information from the provided path + """Extracts issue, valid, lead, and any extra information from the provided path Args: path (str): A path consistent with this PathBuilder @@ -208,36 +262,177 @@ def parse_path(self, path: str) -> dict: Returns: dict: Lookup of all information identified and extracted """ - return self._parse_times(path, self.path_fmt) + # do the core parsing + self._parse_path(path, self.path_fmt) + # return a copy to parsed_args + return deepcopy(self._parsed_args) - def get_issue(self, path: str) -> datetime: + def get_issue(self, path: str) -> datetime | None: """Retrieves the issue date/time from the provided path Args: path (str): A path consistent with this PathBuilder Returns: - datetime: After parsing the path, builds and returns the issue date/time + datetime | None: After parsing the path, builds and returns the issue date/time if + possible else returns None if insufficient info is available to build """ - time_args = self.parse_path(path) - return self.get_issue_from_time_args(time_args) + # do the core parsing + self._parse_path(path, self.path_fmt) + # return a the issue datetime, if determined + return self._parsed_args.get(self.ISSUE, None) - def get_valid(self, path: str) -> datetime: + def get_valid(self, path: str) -> datetime | None: """Retrieves the valid date/time from the provided path Args: path (str): A path consistent with this PathBuilder Returns: - datetime: After parsing the path, builds and returns the valid date/time + datetime | None: After parsing the path, builds and returns the valid date/time if + possible else returns None if insufficient info is available to build + """ + # do the core parsing + self._parse_path(path, self.path_fmt) + # return a the valid datetime, if determined + return self._parsed_args.get(self.VALID, None) + + def _update_lookup(self, fmt_str: str) -> None: + """This method should be called whenever some part of the format has been changed. + + Args: + fmt_str (str): The change format, either part of, or the complete combined, format + + Raises: + ValueError: If the format is not descriptive enough. Formats must specify size and type. """ - time_args = self.parse_path(path) - return self.get_valid_from_time_args(time_args) + # if a format is being updated any cache will be out of date + self._last_parsed_path = None + + for fmt_part in os.path.normpath(fmt_str).split(os.sep): + remaining_fmt_part = fmt_part + lookup_info = [] + cum_start = 0 + while (re_match := re.search(r'\{(.*?)\}', remaining_fmt_part)): + arg_parts = re_match.group()[1:-1].split(':') + if len(arg_parts) != 2: + raise ValueError('Format string must have explicit specification ' + '(must include a ":")') + try: + arg_size = int(re.search(r'^\d+', arg_parts[1]).group()) + except Exception: + # pylint: disable=raise-missing-from + raise ValueError('Format string must have explicit size ' + '(must include a number after ":")') + arg_type = arg_parts[1][-1] + if arg_parts[1][-1] not in [self.INT, self.FLOAT, self.STR]: + raise ValueError('Format string must have explicit type (must include one of ' + f'["{self.INT}", "{self.FLOAT}", "{self.STR}"] after ":")') + + arg_start = re_match.start() + cum_start + arg_end = cum_start = arg_start + arg_size + lookup_info.append(_LookupInfo(arg_parts[0], arg_start, arg_end, arg_type)) + # update the format str to find the next argument + remaining_fmt_part = remaining_fmt_part[re_match.end():] + + exp_len = (sum(end-start for _, start, end, _ in lookup_info) + + len(re.sub(r'\{(.*?)\}', '', fmt_part))) + + self._lookup_dict[fmt_part] = _FormatLookup(exp_len, lookup_info) + # add default for empty string + self._lookup_dict[''] = _FormatLookup(0, []) + + def _parse_path(self, path: str, fmt_str: str) -> None: + """Parse a path for any knowable variables given the provided format string. + + Args: + path (str): The path string to be parsed + fmt_str (str): The format string that the path is assumed to correspond with + """ + if path != self._last_parsed_path: + parsed_arg_parts = self._get_parsed_arg_parts(path, fmt_str) + issue_dt = self._get_issue_from_arg_parts(parsed_arg_parts) + valid_dt = self._get_valid_from_arg_parts(parsed_arg_parts) + # add the issue/valid/lead datetime and timedelta objects + if issue_dt: + parsed_arg_parts[self.ISSUE] = issue_dt + if valid_dt: + parsed_arg_parts[self.VALID] = valid_dt + if issue_dt and valid_dt: + parsed_arg_parts[self.LEAD] = TimeDelta(valid_dt - issue_dt) + + # cache this path and the parsed_arg_parts for repeat requests + self._last_parsed_path = path + self._parsed_args = parsed_arg_parts + + def _get_parsed_arg_parts(self, path: str, fmt_str: str) -> dict: + """Build a dictionary of knowable variable based on path and format string. This + dictionary can be used to create complete issue/valid datetimes and/or contain + extra variables. + + Args: + path (str): The path string from which variables will be extracted + fmt_str (str): The format string used to identify where variables can be found + + Raises: + ValueError: If the path string does not conform to the format string (not expected len) + + Returns: + dict: Dictionary of variables + """ + # Split path and format strings into lists of parts, either dir and/or filenames + fmt_parts = os.path.normpath(fmt_str).split(os.sep) + path_parts = os.path.normpath(path).split(os.sep) + + parsed_arg_parts = {} + for path_part, fmt_part in zip(path_parts, fmt_parts): + expected_len, lookup_info = self._lookup_dict[fmt_part] + if (part_len := len(path_part)) != expected_len: + raise ValueError('Path is not expected length. Passed path part ' + f"'{path_part}' doesn't match format '{fmt_part}'") + for lookup in lookup_info: + if not (0 <= lookup.start <= part_len and 0 <= lookup.end <= part_len): + raise ValueError('Parse indices are out of range for path') + try: + match lookup.type: + case self.INT: + parsed_arg_parts[lookup.key] = int(path_part[lookup.start:lookup.end]) + case self.FLOAT: + parsed_arg_parts[lookup.key] = float(path_part[lookup.start:lookup.end]) + case self.STR: + parsed_arg_parts[lookup.key] = path_part[lookup.start:lookup.end] + except ValueError as exc: + raise ValueError('Unable to apply formatting') from exc + return parsed_arg_parts + + def _apply_format(self, fmt_str: str, **kwargs) -> str: + """Use the format string and any variables in the kwargs to build a path. + + Args: + fmt_str (str): A format string, for part or a whole path + + Raises: + ValueError: If the generated path part does not match expected length + + Returns: + str: A string representation of a system path, combined with os specific separators + """ + path_parts = [] + # we split the format string without normalizing to maintain user specified path + # struct such as a double separator (sometime this can be needed) + for fmt_part in fmt_str.split(os.sep): + path_part = fmt_part.format_map(kwargs) + if len(path_part) == self._lookup_dict[fmt_part].exp_len: + path_parts.append(path_part) + else: + raise ValueError('Arguments generate a path that violate ' + f"at least part of the format, part '{fmt_part}'") + return os.path.sep.join(path_parts) @staticmethod - def get_issue_from_time_args(parsed_args: dict, - valid: datetime | None = None, - lead: timedelta | None = None) -> datetime: + def _get_issue_from_arg_parts(parsed_args: dict, + valid: datetime | None = None, + lead: timedelta | None = None) -> datetime: """Static method for creating an issue date/time from parsed arguments and optional inputs Args: @@ -260,22 +455,18 @@ def get_issue_from_time_args(parsed_args: dict, parsed_args.get('issue.second', 0), parsed_args.get('issue.microsecond', 0), tzinfo=UTC) - if lead is None and 'lead.hour' in parsed_args: - lead = PathBuilder.get_lead_from_time_args(parsed_args) - + lead = PathBuilder._get_lead_from_time_args(parsed_args) if valid is None and 'valid.year' in parsed_args: - valid = PathBuilder.get_valid_from_time_args(parsed_args) - + valid = PathBuilder._get_valid_from_arg_parts(parsed_args) if valid and lead: return valid - lead - return None @staticmethod - def get_valid_from_time_args(parsed_args: dict, - issue: datetime | None = None, - lead: timedelta | None = None) -> datetime: + def _get_valid_from_arg_parts(parsed_args: dict, + issue: datetime | None = None, + lead: timedelta | None = None) -> datetime: """Static method for creating a valid date/time from parsed arguments and optional inputs Args: @@ -298,20 +489,16 @@ def get_valid_from_time_args(parsed_args: dict, parsed_args.get('valid.second', 0), parsed_args.get('valid.microsecond', 0), tzinfo=UTC) - if lead is None and 'lead.hour' in parsed_args: - lead = PathBuilder.get_lead_from_time_args(parsed_args) - + lead = PathBuilder._get_lead_from_time_args(parsed_args) if issue is None and 'issue.year' in parsed_args: - issue = PathBuilder.get_issue_from_time_args(parsed_args) - + issue = PathBuilder._get_issue_from_arg_parts(parsed_args) if issue and lead: return issue + lead - return None @staticmethod - def get_lead_from_time_args(time_args: dict) -> timedelta: + def _get_lead_from_time_args(time_args: dict) -> timedelta: """Static method for creating a lead time from parsed arguments and optional inputs Args: @@ -323,13 +510,22 @@ def get_lead_from_time_args(time_args: dict) -> timedelta: """ if 'lead.hour' in time_args.keys(): return timedelta(hours=time_args['lead.hour']) - return None @staticmethod - def _ensure_lead(issue: datetime, - valid: datetime, - lead: timedelta | TimeDelta) -> TimeDelta: + def _ensure_lead(issue: datetime | None, + valid: datetime | None, + lead: timedelta | TimeDelta | None) -> TimeDelta: + """Make every attempt to ensure lead is known, by calculating or converting if needed. + + Args: + issue (datetime | None): An issue datetime if known, else None + valid (datetime | None): A valid datetime if known, else None + lead (timedelta | TimeDelta | None): A lead if known, else None + + Returns: + TimeDelta: _description_ + """ if lead: if isinstance(lead, timedelta): return TimeDelta(lead) @@ -338,29 +534,17 @@ def _ensure_lead(issue: datetime, return TimeDelta(valid-issue) return None - def _parse_times(self, string: str, format_str: str) -> dict: - def parse_args(key: str, value: str, result: dict): - for arg in key.split('{')[1:]: - var_name, var_size = arg.split(':') - var_type = var_size[2:3] - var_size = int(var_size[0:2]) - match var_type: - case 'd': - result[var_name] = int(value[:var_size]) - case _: - raise ValueError(f'Unknown format type: {var_type}') - key = key[var_size:] - # Check for additional characters following the end of the format element to reach - # next offset position for value... - value = value[var_size + len(arg.partition('}')[2]):] - - # Update to more generically handle time formats... - dirs = os.path.normpath(format_str).split(os.sep) - vals = os.path.normpath(string).split(os.sep) - time_args = {} - for i, _dir in enumerate(dirs): - res = re.search(r'{.*}', _dir) - if res: - parse_args(res.group(), vals[i][res.span()[0]:], time_args) - - return time_args + +# Private utility classes +class _LookupInfo(NamedTuple): + """Data class used to hold lookup info""" + key: str + start: int + end: int + type: str # should be one of 'd', 'f', 's' + + +class _FormatLookup(NamedTuple): + """Data class used to hold format and lookup info""" + exp_len: int + lookups: list[_LookupInfo] diff --git a/python/idsse_common/idsse/common/rabbitmq_utils.py b/python/idsse_common/idsse/common/rabbitmq_utils.py index 770a2907..8887829c 100644 --- a/python/idsse_common/idsse/common/rabbitmq_utils.py +++ b/python/idsse_common/idsse/common/rabbitmq_utils.py @@ -92,7 +92,9 @@ class RabbitMqParamsAndCallback(NamedTuple): class RabbitMqMessage(NamedTuple): - """Data class to hold a RabbitMQ message body, properties, and optional route_key (if outbound)""" + """ + Data class to hold a RabbitMQ message body, properties, and optional route_key (if outbound) + """ body: str properties: BasicProperties route_key: str | None = None @@ -143,13 +145,15 @@ def __init__( self._consumer_tags.append( self.channel.basic_consume(queue.name, partial(self._on_message, func=func), - # RMQ requires auto_ack=True to consume from Direct Reply-to + # RMQ requires auto_ack=True for Direct Reply-to auto_ack=queue.name == DIRECT_REPLY_QUEUE) ) def run(self): _set_context(self.context) - logger.info('Start Consuming... (to stop press CTRL+C)') + # create a local logger since this is run in a separate threat when start() is called + _logger = logging.getLogger(f'{__name__}::{self.__class__.__name__}') + _logger.info('Start Consuming... (to stop press CTRL+C)') self.channel.start_consuming() def stop(self): @@ -221,7 +225,7 @@ def __init__( 'x-message-ttl': 10 * 1000}) _setup_exch_and_queue(self.channel, self._exch, self._queue) - elif self._exch.name != '': # if using default exchange, skip declaring (not allowed by RMQ) + elif self._exch.name != '': # if using default exchange, skip declare (not allowed by RMQ) _setup_exch(self.channel, self._exch) if self._exch.delivery_conf: @@ -229,7 +233,9 @@ def __init__( def run(self): _set_context(self.context) - logger.info('Starting publisher') + # create a local logger since this is run in a separate threat when start() is called + _logger = logging.getLogger(f'{__name__}::{self.__class__.__name__}') + _logger.info('Starting publisher') while self._is_running: if self.connection and self.connection.is_open: self.connection.process_data_events(time_limit=1) @@ -405,8 +411,8 @@ def _response_callback( body: bytes ): """Handle RabbitMQ message emitted to response queue.""" - logger.debug('Received response message with routing_key: %s, content_type: %s, message: %i', - method.routing_key, properties.content_type, str(body, encoding='utf-8')) + logger.debug('Received response with routing_key: %s, content_type: %s, message: %i', + method.routing_key, properties.content_type, str(body, encoding='utf-8')) # remove future from pending list. we will update result shortly request_future = self._pending_requests.pop(properties.correlation_id) @@ -560,8 +566,6 @@ def threadsafe_nack( threadsafe_call(channel, lambda: channel.basic_nack(delivery_tag, requeue=requeue)) - - def _initialize_exchange_and_queue(channel: Channel, params: RabbitMqParams) -> str: """Declare and bind RabbitMQ exchange and queue using the provided channel. @@ -634,7 +638,7 @@ def _setup_exch_and_queue(channel: Channel, exch: Exch, queue: Queue): queue.arguments['x-queue-type'] == 'quorum' and queue.auto_delete: raise ValueError('Quorum queues can not be configured to auto delete') - if exch.name != '': # if using default exchange, skip declaring (not allowed by RMQ) + if exch.name != '': # if using default exchange, skip declaring (not allowed by RMQ) _setup_exch(channel, exch) if queue.name == DIRECT_REPLY_QUEUE: @@ -652,7 +656,7 @@ def _setup_exch_and_queue(channel: Channel, exch: Exch, queue: Queue): queue_name = result.method.queue logger.debug('Declared queue: %s', queue_name) - if exch.name != '': # if using default exchange, skip binding queues (not allowed by RMQ) + if exch.name != '': # if using default exchange, skip binding queues (not allowed by RMQ) if isinstance(queue.route_key, list): for route_key in queue.route_key: channel.queue_bind( @@ -750,7 +754,6 @@ def _blocking_publish( channel (BlockingChannel): the pika channel to use to publish. exch (Exch): parameters for the RabbitMQ exchange to publish message to. message_params (RabbitMqMessage): the message body to publish, plus properties and - (optional) route_key queue (optional, Queue | None): parameters for RabbitMQ queue, if message is being published to a "temporary"/"private" message queue. The published message will be purged from this queue after its TTL expires. diff --git a/python/idsse_common/idsse/common/utils.py b/python/idsse_common/idsse/common/utils.py index c35e97de..97995553 100644 --- a/python/idsse_common/idsse/common/utils.py +++ b/python/idsse_common/idsse/common/utils.py @@ -31,26 +31,37 @@ class RoundingMethod(Enum): RoundingParam = str | RoundingMethod -class TimeDelta: - """Wrapper class for datetime.timedelta to add helpful properties""" - - def __init__(self, time_delta: timedelta) -> None: - self._td = time_delta +class TimeDelta(timedelta): + """Extend class for datetime.timedelta to add helpful properties.""" + def __new__(cls, *args, **kwargs): + if isinstance(args[0], timedelta): + return super().__new__(cls, seconds=args[0].total_seconds()) + return super().__new__(cls, *args, **kwargs) @property def minute(self): """Property to get the number of minutes this instance represents""" - return int(self._td / timedelta(minutes=1)) + return int(self / timedelta(minutes=1)) + + @property + def minutes(self): + """Property to get the number of minutes this instance represents""" + return self.minute @property def hour(self): """Property to get the number of hours this instance represents""" - return int(self._td / timedelta(hours=1)) + return int(self / timedelta(hours=1)) + + @property + def hours(self): + """Property to get the number of hours this instance represents""" + return self.hour @property def day(self): """Property to get the number of days this instance represents""" - return self._td.days + return self.days class Map(dict): @@ -200,6 +211,7 @@ def _round_toward_zero(number: float) -> int: func = math.ceil if number < 0 else math.floor return func(number) + def round_half_away(number: int | float, precision: int = 0) -> int | float: """ *Deprecated: avoid using this function directly, instead use idsse.commons.round_()* diff --git a/python/idsse_common/test/test_path_builder.py b/python/idsse_common/test/test_path_builder.py index 7843199e..7b0dfb87 100644 --- a/python/idsse_common/test/test_path_builder.py +++ b/python/idsse_common/test/test_path_builder.py @@ -9,34 +9,11 @@ # # -------------------------------------------------------------------------------- # pylint: disable=missing-function-docstring,invalid-name,redefined-outer-name,protected-access -# cspell:words pathbuilder -from datetime import datetime, timedelta, UTC -import pytest - -from idsse.common.utils import TimeDelta -from idsse.common.path_builder import PathBuilder - - -def test_from_dir_filename_creates_valid_pathbuilder(): - directory = './test_directory' - filename = 'some_file.txt' - path_builder = PathBuilder.from_dir_filename(directory, filename) - - assert isinstance(path_builder, PathBuilder) - assert path_builder._basedir == directory - assert path_builder._file_ext == '' - - -def test_from_path_creates_valid_pathbuilder(): - base_dir = './test_directory' - path_builder = PathBuilder.from_path(f'{base_dir}/some_file.txt') - - assert isinstance(path_builder, PathBuilder) - assert path_builder._basedir == base_dir - assert path_builder._file_base == base_dir - assert path_builder._file_ext == '' +from datetime import datetime, UTC +from pytest import fixture, raises +from idsse.common.path_builder import TimeDelta, PathBuilder # properties EXAMPLE_BASE_DIR = './some/directory' @@ -44,13 +21,56 @@ def test_from_path_creates_valid_pathbuilder(): EXAMPLE_FILE = 'my_file' EXAMPLE_FILE_EXT = '.txt' +EXAMPLE_ISSUE = datetime(1970, 10, 3, 12, tzinfo=UTC) # a.k.a. issued at +EXAMPLE_VALID = datetime(1970, 10, 3, 14, tzinfo=UTC) # a.k.a. valid until +EXAMPLE_LEAD = TimeDelta(EXAMPLE_VALID - EXAMPLE_ISSUE) # a.k.a. duration of time that issue lasts +EXAMPLE_FULL_PATH = '~/blend.19701003/12/core/blend.t12z.core.f002.co.grib2.idx' -@pytest.fixture + +@fixture def local_path_builder() -> PathBuilder: # create example Pa†hBuilder instance using test strings return PathBuilder(EXAMPLE_BASE_DIR, EXAMPLE_SUB_DIR, EXAMPLE_FILE, EXAMPLE_FILE_EXT) +@fixture +def path_builder() -> PathBuilder: + subdirectory_pattern = ( + 'blend.{issue.year:04d}{issue.month:02d}{issue.day:02d}/{issue.hour:02d}/core/' + ) + file_base_pattern = 'blend.t{issue.hour:02d}z.core.f{lead.hour:03d}.co' + return PathBuilder('~', subdirectory_pattern, file_base_pattern, 'grib2.idx') + + +@fixture +def path_builder_with_region() -> PathBuilder: + subdirectory_pattern = ( + 'blend.{issue.year:04d}{issue.month:02d}{issue.day:02d}/{issue.hour:02d}/core/' + ) + file_base_pattern = 'blend.t{issue.hour:02d}z.core.f{lead.hour:03d}.{region:2s}' + return PathBuilder('~', subdirectory_pattern, file_base_pattern, 'grib2.idx') + + +def test_from_dir_filename_creates_valid_path_builder(): + directory = './test_directory' + filename = 'some_file.txt' + path_builder = PathBuilder.from_dir_filename(directory, filename) + + assert isinstance(path_builder, PathBuilder) + assert path_builder.base_dir == directory + assert path_builder.file_ext == '.txt' + + +def test_from_path_creates_valid_path_builder(): + base_dir = './test_directory' + filename = 'some_file.txt' + path_builder = PathBuilder.from_path(f'{base_dir}/{filename}') + assert isinstance(path_builder, PathBuilder) + assert path_builder.base_dir == base_dir + assert path_builder.file_base == filename + assert path_builder.file_ext == '.txt' + + def test_dir_fmt(local_path_builder: PathBuilder): assert local_path_builder.dir_fmt == f'{EXAMPLE_BASE_DIR}/{EXAMPLE_SUB_DIR}' @@ -69,31 +89,14 @@ def test_path_fmt(local_path_builder: PathBuilder): ) -# methods -EXAMPLE_ISSUE = datetime(1970, 10, 3, 12, tzinfo=UTC) # a.k.a. issued at -EXAMPLE_VALID = datetime(1970, 10, 3, 14, tzinfo=UTC) # a.k.a. valid until -EXAMPLE_LEAD = TimeDelta(EXAMPLE_VALID - EXAMPLE_ISSUE) # a.k.a. duration of time that issue lasts - -EXAMPLE_FULL_PATH = '~/blend.19701003/12/core/blend.t12z.core.f002.co.grib2.idx' - - -@pytest.fixture -def path_builder() -> PathBuilder: - subdirectory_pattern = ( - 'blend.{issue.year:04d}{issue.month:02d}{issue.day:02d}/{issue.hour:02d}/core/' - ) - file_base_pattern = 'blend.t{issue.hour:02d}z.core.f{lead.hour:03d}.co' - return PathBuilder('~', subdirectory_pattern, file_base_pattern, 'grib2.idx') - - def test_build_dir_gets_issue_valid_and_lead(path_builder: PathBuilder): - result_dict = path_builder.build_dir(issue=EXAMPLE_ISSUE) - assert result_dict == '~/blend.19701003/12/core/' + result = path_builder.build_dir(issue=EXAMPLE_ISSUE) + assert result == '~/blend.19701003/12/core/' def test_build_dir_fails_without_issue(path_builder: PathBuilder): - result_dict = path_builder.build_dir(issue=None) - assert result_dict is None + result = path_builder.build_dir(issue=None) + assert result is None def test_build_filename(path_builder: PathBuilder): @@ -106,6 +109,38 @@ def test_build_path(path_builder: PathBuilder): assert result_filepath == '~/blend.19701003/12/core/blend.t12z.core.f002.co.grib2.idx' +def test_build_path_with_invalid_lead(path_builder: PathBuilder): + # if lead needs more than 3 chars to be represented, ValueError will be raised + with raises(ValueError): + path_builder.build_path(issue=EXAMPLE_ISSUE, + lead=EXAMPLE_LEAD*1000) + + +def test_build_path_with_region(path_builder_with_region: PathBuilder): + region = 'co' + result = path_builder_with_region.build_path(issue=EXAMPLE_ISSUE, + lead=EXAMPLE_LEAD, + region=region) + result_dict = path_builder_with_region.parse_path(result) + assert result_dict['issue'] == EXAMPLE_ISSUE + assert result_dict['lead'] == EXAMPLE_LEAD + assert result_dict['region'] == region + + +def test_build_path_with_invalid_region(path_builder_with_region: PathBuilder): + # if region is more than 2 chars, ValueError will be raised + with raises(ValueError): + path_builder_with_region.build_path(issue=EXAMPLE_ISSUE, + lead=EXAMPLE_LEAD, + region='conus') + + +def test_build_path_with_required_but_missing_region(path_builder_with_region: PathBuilder): + # if a required variable (region) is not provided, KeyError will be raised + with raises(KeyError): + path_builder_with_region.build_path(issue=EXAMPLE_ISSUE, lead=EXAMPLE_LEAD) + + def test_parse_dir(path_builder: PathBuilder): result_dict = path_builder.parse_dir(EXAMPLE_FULL_PATH) @@ -137,57 +172,3 @@ def test_get_valid_returns_none_when_issue_or_lead_failed(path_builder: PathBuil result_valid = path_builder.get_valid(path_with_invalid_lead) assert result_valid is None - - -# static methods -def test_get_issue_from_time_args(path_builder: PathBuilder): - parsed_dict = path_builder.parse_path(EXAMPLE_FULL_PATH) - issue_result = PathBuilder.get_issue_from_time_args(parsed_args=parsed_dict) - - assert issue_result == EXAMPLE_ISSUE - - -def test_get_issue_returns_none_if_args_empty(): - issue_result = PathBuilder.get_issue_from_time_args({}) - assert issue_result is None - - -def test_get_valid_from_time_args(): - parsed_dict = {} - parsed_dict['valid.year'] = 1970 - parsed_dict['valid.month'] = 10 - parsed_dict['valid.day'] = 3 - parsed_dict['valid.hour'] = 14 - - valid_result = PathBuilder.get_valid_from_time_args(parsed_dict) - assert valid_result == EXAMPLE_VALID - - -def test_get_valid_returns_none_if_args_empty(): - valid_result = PathBuilder.get_valid_from_time_args({}) - assert valid_result is None - - -def test_get_valid_from_time_args_calculates_based_on_lead(path_builder: PathBuilder): - parsed_dict = path_builder.parse_path(EXAMPLE_FULL_PATH) - result_valid: datetime = PathBuilder.get_valid_from_time_args(parsed_args=parsed_dict) - assert result_valid == EXAMPLE_VALID - - -def test_get_lead_from_time_args(path_builder: PathBuilder): - parsed_dict = path_builder.parse_path(EXAMPLE_FULL_PATH) - lead_result: timedelta = PathBuilder.get_lead_from_time_args(parsed_dict) - assert lead_result.seconds == EXAMPLE_LEAD.minute * 60 - - -def test_calculate_issue_from_valid_and_lead(): - parsed_dict = { - 'valid.year': 1970, - 'valid.month': 10, - 'valid.day': 3, - 'valid.hour': 14, - 'lead.hour': 2 - } - - result_issue = PathBuilder.get_issue_from_time_args(parsed_args=parsed_dict) - assert result_issue == EXAMPLE_ISSUE