Skip to content
Snippets Groups Projects
Select Git revision
  • 1b886c1f275fbf562106261e3602a2add7536865
  • MCCS-163 default
  • main
  • sar-277-update-docs-with-examples-for-lrc
  • st-946-automate
  • sar_302-log-fix
  • sar-287_subarray_commands_to_lrc
  • sar_302-POC_await_sub_device_state
  • sat_302_fix_pipelines
  • sar-286_lrc_one_subarry_command
  • sar-286_lrc_improvements
  • sar-288-async-controller
  • sar-276-combine-tango-queue
  • sar-255_remove_nexus_reference
  • sar-275-add-LRC
  • sar-273-add-lrc-attributes
  • sar-272
  • sp-1106-marvin-1230525148-ska-tango-base
  • sp-1106-marvin-813091765-ska-tango-base
  • sar-255/Publish-package-to-CAR
  • mccs-661-device-under-test-fixture
  • mccs-659-pep257-docstring-linting
  • 0.11.3
  • 0.11.2
  • 0.11.1
  • 0.11.0
  • 0.10.1
  • 0.10.0
  • 0.9.1
  • 0.9.0
  • 0.8.1
  • 0.8.0
  • 0.7.2
  • 0.7.1
  • 0.7.0
  • 0.6.6
  • 0.6.5
  • 0.6.4
  • 0.6.3
  • 0.6.2
  • 0.6.1
  • 0.6.0
42 results

conf.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    datetimeutils.py 7.10 KiB
    #  Copyright (C) 2012 ASTRON (Netherlands Institute for Radio Astronomy)
    #  SPDX-License-Identifier: GPL-3.0-or-later
    
    from datetime import datetime, timedelta, timezone
    
    
    def monthRanges(min_date: datetime, max_date: datetime, month_step: int=1):
        ranges = []
    
        min_month_start = datetime(min_date.year, min_date.month, 1, tzinfo=min_date.tzinfo)
    
        month_start = min_month_start
        while month_start < max_date:
            if month_start.month <= 12-month_step:
                month_end = datetime(month_start.year, month_start.month+month_step, 1, tzinfo=month_start.tzinfo) - timedelta(milliseconds=1)
            else:
                month_end = datetime(month_start.year+1, month_start.month-12+month_step, 1, tzinfo=month_start.tzinfo) - timedelta(milliseconds=1)
    
            ranges.append((month_start, month_end))
    
            if month_start.month <= 12-month_step:
                month_start = datetime(month_start.year, month_start.month+month_step, 1, tzinfo=min_date.tzinfo)
            else:
                month_start = datetime(month_start.year+1, month_start.month-12+month_step, 1, tzinfo=min_date.tzinfo)
    
        return ranges
    
    def totalSeconds(td_value) -> float:
        '''Return the total number of fractional seconds contained in the duration.
        For Python < 2.7 compute it, else use total_seconds() method.
        '''
        if hasattr(td_value,"total_seconds"):
            return td_value.total_seconds()
    
        return (td_value.microseconds + (td_value.seconds + td_value.days * 86400) * 1000000) / 1000000.0
    
    def format_timedelta(td: timedelta) -> str:
        '''Return string representation of timedelta value td, which works even for negative values.
        Normal python is weird: str(timedelta(hours=-1)) becomes '-1 day, 23:00:00'
        With this function: format_timedelta(timedelta(hours=-1)) becomes '-1:00:00' which makes much more sense!
        '''
        if td < timedelta(0):
            return '-' + str(-td)
        return str(td)
    
    def parseDatetime(date_time: str) -> datetime:
        """ Parse the datetime format used in LOFAR parsets. """
        return datetime.strptime(date_time, ('%Y-%m-%d %H:%M:%S.%f' if '.' in date_time else '%Y-%m-%d %H:%M:%S')).replace(tzinfo=timezone.utc) # LOFAR uses UTC
    
    def formatDatetime(timestamp: datetime) -> str:
        """ Format the timestamp as used in LOFAR parsets. """
        try:
            return timestamp.strftime(('%Y-%m-%d %H:%M:%S' if timestamp.microsecond == 0 else '%Y-%m-%d %H:%M:%S.%f'))
        except AttributeError:
            return str(None)
    
    MDJ_EPOCH = datetime(1858, 11, 17, 0, 0, 0)
    
    def to_modified_julian_date(timestamp: datetime) -> float:
        '''
        computes the modified_julian_date from a python datetime timestamp
        :param timestamp: datetime a python datetime timestamp
        :return: double, the modified_julian_date
        '''
        return to_modified_julian_date_in_seconds(timestamp)/86400.0
    
    def to_modified_julian_date_in_seconds(timestamp: datetime) -> float:
        '''
        computes the modified_julian_date (in seconds as opposed to the official days) from a python datetime timestamp
        :param timestamp: datetime a python datetime timestamp
        :return: double, the modified_julian_date (fractional number of seconds since MJD_EPOCH)
        '''
        return totalSeconds(timestamp - MDJ_EPOCH)
    
    def from_modified_julian_date(modified_julian_date: float) -> datetime:
        '''
        computes the python datetime timestamp from a modified_julian_date
        :param modified_julian_date: double, a timestamp expressed in modified_julian_date format (fractional number of days since MJD_EPOCH)
        :return: datetime, the timestamp as python datetime
        '''
        return from_modified_julian_date_in_seconds(modified_julian_date*86400.0)
    
    def from_modified_julian_date_in_seconds(modified_julian_date_secs: float) -> datetime:
        '''
        computes the python datetime timestamp from a modified_julian_date (in seconds as opposed to the official days)
        :param modified_julian_date: double, a timestamp expressed in modified_julian_date format (fractional number of seconds since MJD_EPOCH)
        :return: datetime, the timestamp as python datetime
        '''
        return MDJ_EPOCH + timedelta(seconds=modified_julian_date_secs)
    
    def to_seconds_since_unix_epoch(timestamp: datetime) -> float:
        '''
        computes the (fractional) number of seconds since the unix epoch for a python datetime.timestamp
        :param timestamp: datetime a python datetime timestamp (in UTC)
        :return: double, the (fractional) number of seconds since the unix epoch
        '''
        return totalSeconds(timestamp - datetime.fromtimestamp(0, tz=timezone.utc))
    
    def to_milliseconds_since_unix_epoch(timestamp: datetime) -> float:
        '''
        computes the (fractional) number of milliseconds since the unix epoch for a python datetime.timestamp
        :param timestamp: datetime a python datetime timestamp
        :return: double, the (fractional) number of milliseconds since the unix epoch
        '''
        return 1000.0 * to_seconds_since_unix_epoch(timestamp)
    
    def from_seconds_since_unix_epoch(nr_of_seconds_since_epoch: float) -> datetime:
        '''
        computes a python datetime.timestamp given the (fractional) number of seconds since the unix epoch
        :param double or int, the (fractional) number of seconds since the unix epoch
        :return: timestamp: datetime a python datetime timestamp (in UTC)
        '''
        return datetime.fromtimestamp(nr_of_seconds_since_epoch, tz=timezone.utc)
    
    def from_milliseconds_since_unix_epoch(nr_of_milliseconds_since_epoch: float) -> datetime:
        '''
        computes a python datetime.timestamp given the (fractional) number of milliseconds since the unix epoch
        :param double or int, the (fractional) number of milliseconds since the unix epoch
        :return: timestamp: datetime a python datetime timestamp (in UTC)
        '''
        return from_seconds_since_unix_epoch(nr_of_milliseconds_since_epoch/1000.0)
    
    def round_to_millisecond_precision(timestamp: datetime) -> datetime:
        """
        returns the given timestamp rounded to the nearest millisecond
        :param timestamp: datetime a python datetime timestamp
        :return: the given timestamp rounded to the nearest millisecond
        """
        diff_to_rounded_millisecond = timestamp.microsecond - 1000*round(timestamp.microsecond/1000)
        return timestamp - timedelta(microseconds=diff_to_rounded_millisecond)
    
    def round_to_second_precision(timestamp: datetime) -> datetime:
        """
        returns the given timestamp rounded to the nearest second
        :param timestamp: datetime a python datetime timestamp
        :return: the given timestamp rounded to the nearest second
        """
        if timestamp.microsecond < 500000:
            return timestamp + timedelta(microseconds=-timestamp.microsecond)
        else:
            return timestamp + timedelta(microseconds=-timestamp.microsecond, seconds=1)
    
    def round_to_minute_precision(timestamp: datetime) -> datetime:
        """
        returns the given timestamp rounded to the nearest minute
        :param timestamp: datetime a python datetime timestamp
        :return: the given timestamp rounded to the nearest minute
        """
        if timestamp.second < 30:
            return timestamp + timedelta(seconds=-timestamp.second, microseconds=-timestamp.microsecond)
        else:
            return timestamp + timedelta(minutes=1, seconds=-timestamp.second, microseconds=-timestamp.microsecond)