sdks/python/apache_beam/utils/timestamp.py - beam - Git at Google

 #
 # Licensed to the Apache Software Foundation (ASF) under one or more
 # contributor license agreements.  See the NOTICE file distributed with
 # this work for additional information regarding copyright ownership.
 # The ASF licenses this file to You under the Apache License, Version 2.0
 # (the "License"); you may not use this file except in compliance with
 # the License.  You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #

 """Timestamp utilities.

 For internal use only; no backwards-compatibility guarantees.
 """

 from __future__ import absolute_import
 from __future__ import division

 import datetime
 import functools
 import re
 from builtins import object

 import pytz
 from past.builtins import long

 from apache_beam.portability import common_urns


 @functools.total_ordering
 class Timestamp(object):
   """Represents a Unix second timestamp with microsecond granularity.

   Can be treated in common timestamp arithmetic operations as a numeric type.

   Internally stores a time interval as an int of microseconds. This strategy
   is necessary since floating point values lose precision when storing values,
   especially after arithmetic operations (for example, 10000000 % 0.1 evaluates
   to 0.0999999994448885).
   """

   def __init__(self, seconds=0, micros=0):
     if not isinstance(seconds, (int, long, float)):
       raise TypeError('Cannot interpret %s %s as seconds.' % (
           seconds, type(seconds)))
     if not isinstance(micros, (int, long, float)):
       raise TypeError('Cannot interpret %s %s as micros.' % (
           micros, type(micros)))
     self.micros = int(seconds * 1000000) + int(micros)

   @staticmethod
   def of(seconds):
     """Return the Timestamp for the given number of seconds.

     If the input is already a Timestamp, the input itself will be returned.

     Args:
       seconds: Number of seconds as int, float, long, or Timestamp.

     Returns:
       Corresponding Timestamp object.
     """

     if not isinstance(seconds, (int, long, float, Timestamp)):
       raise TypeError('Cannot interpret %s %s as Timestamp.' % (
           seconds, type(seconds)))
     if isinstance(seconds, Timestamp):
       return seconds
     return Timestamp(seconds)

   RFC_3339_RE = re.compile(
       r'^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.(\d+))?Z$')

   @staticmethod
   def _epoch_datetime_utc():
     return datetime.datetime.fromtimestamp(0, pytz.utc)

   @classmethod
   def from_utc_datetime(cls, dt):
     """Create a ``Timestamp`` instance from a ``datetime.datetime`` object.

     Args:
       dt: A ``datetime.datetime`` object in UTC (offset-aware).
     """
     if dt.tzinfo != pytz.utc:
       raise ValueError('dt not in UTC: %s' % dt)
     duration = dt - cls._epoch_datetime_utc()
     return Timestamp(duration.total_seconds())

   @classmethod
   def from_rfc3339(cls, rfc3339):
     """Create a ``Timestamp`` instance from an RFC 3339 compliant string.

     Args:
       rfc3339: String in RFC 3339 form.
     """
     dt_args = []
     match = cls.RFC_3339_RE.match(rfc3339)
     if match is None:
       raise ValueError('Could not parse RFC 3339 string: %s' % rfc3339)
     for s in match.groups():
       if s is not None:
         dt_args.append(int(s))
       else:
         dt_args.append(0)
     dt_args += (pytz.utc, )
     dt = datetime.datetime(*dt_args)
     return cls.from_utc_datetime(dt)

   def predecessor(self):
     """Returns the largest timestamp smaller than self."""
     return Timestamp(micros=self.micros - 1)

   def __repr__(self):
     micros = self.micros
     sign = ''
     if micros < 0:
       sign = '-'
       micros = -micros
     int_part = micros // 1000000
     frac_part = micros % 1000000
     if frac_part:
       return 'Timestamp(%s%d.%06d)' % (sign, int_part, frac_part)
     return 'Timestamp(%s%d)' % (sign, int_part)

   def to_utc_datetime(self):
     # We can't easily construct a datetime object from microseconds, so we
     # create one at the epoch and add an appropriate timedelta interval.
     return self._epoch_datetime_utc().replace(tzinfo=None) + datetime.timedelta(
         microseconds=self.micros)

   def to_rfc3339(self):
     # Append 'Z' for UTC timezone.
     return self.to_utc_datetime().isoformat() + 'Z'

   def __float__(self):
     # Note that the returned value may have lost precision.
     return self.micros / 1000000

   def __int__(self):
     # Note that the returned value may have lost precision.
     return self.micros // 1000000

   def __eq__(self, other):
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Duration):
       other = Timestamp.of(other)
     return self.micros == other.micros

   def __ne__(self, other):
     # TODO(BEAM-5949): Needed for Python 2 compatibility.
     return not self == other

   def __lt__(self, other):
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Duration):
       other = Timestamp.of(other)
     return self.micros < other.micros

   def __hash__(self):
     return hash(self.micros)

   def __add__(self, other):
     other = Duration.of(other)
     return Timestamp(micros=self.micros + other.micros)

   def __radd__(self, other):
     return self + other

   def __sub__(self, other):
     other = Duration.of(other)
     return Timestamp(micros=self.micros - other.micros)

   def __mod__(self, other):
     other = Duration.of(other)
     return Duration(micros=self.micros % other.micros)


 MIN_TIMESTAMP = Timestamp(micros=int(
     common_urns.constants.MIN_TIMESTAMP_MILLIS.constant)*1000)
 MAX_TIMESTAMP = Timestamp(micros=int(
     common_urns.constants.MAX_TIMESTAMP_MILLIS.constant)*1000)


 @functools.total_ordering
 class Duration(object):
   """Represents a second duration with microsecond granularity.

   Can be treated in common arithmetic operations as a numeric type.

   Internally stores a time interval as an int of microseconds. This strategy
   is necessary since floating point values lose precision when storing values,
   especially after arithmetic operations (for example, 10000000 % 0.1 evaluates
   to 0.0999999994448885).
   """

   def __init__(self, seconds=0, micros=0):
     self.micros = int(seconds * 1000000) + int(micros)

   @staticmethod
   def of(seconds):
     """Return the Duration for the given number of seconds since Unix epoch.

     If the input is already a Duration, the input itself will be returned.

     Args:
       seconds: Number of seconds as int, float or Duration.

     Returns:
       Corresponding Duration object.
     """

     if isinstance(seconds, Timestamp):
       raise TypeError('Cannot interpret %s as Duration.' % seconds)
     if isinstance(seconds, Duration):
       return seconds
     return Duration(seconds)

   def __repr__(self):
     micros = self.micros
     sign = ''
     if micros < 0:
       sign = '-'
       micros = -micros
     int_part = micros // 1000000
     frac_part = micros % 1000000
     if frac_part:
       return 'Duration(%s%d.%06d)' % (sign, int_part, frac_part)
     return 'Duration(%s%d)' % (sign, int_part)

   def __float__(self):
     # Note that the returned value may have lost precision.
     return self.micros / 1000000

   def __eq__(self, other):
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Timestamp):
       other = Duration.of(other)
     return self.micros == other.micros

   def __ne__(self, other):
     # TODO(BEAM-5949): Needed for Python 2 compatibility.
     return not self == other

   def __lt__(self, other):
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Timestamp):
       other = Duration.of(other)
     return self.micros < other.micros

   def __hash__(self):
     return hash(self.micros)

   def __neg__(self):
     return Duration(micros=-self.micros)

   def __add__(self, other):
     if isinstance(other, Timestamp):
       return other + self
     other = Duration.of(other)
     return Duration(micros=self.micros + other.micros)

   def __radd__(self, other):
     return self + other

   def __sub__(self, other):
     other = Duration.of(other)
     return Duration(micros=self.micros - other.micros)

   def __rsub__(self, other):
     return -(self - other)

   def __mul__(self, other):
     other = Duration.of(other)
     return Duration(micros=self.micros * other.micros // 1000000)

   def __rmul__(self, other):
     return self * other

   def __mod__(self, other):
     other = Duration.of(other)
     return Duration(micros=self.micros % other.micros)


 # The minimum granularity / interval expressible in a Timestamp / Duration
 # object.
 TIME_GRANULARITY = Duration(micros=1)
	#
	# Licensed to the Apache Software Foundation (ASF) under one or more
	# contributor license agreements. See the NOTICE file distributed with
	# this work for additional information regarding copyright ownership.
	# The ASF licenses this file to You under the Apache License, Version 2.0
	# (the "License"); you may not use this file except in compliance with
	# the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.
	#

	"""Timestamp utilities.

	For internal use only; no backwards-compatibility guarantees.
	"""

	from __future__ import absolute_import
	from __future__ import division

	import datetime
	import functools
	import re
	from builtins import object

	import pytz
	from past.builtins import long

	from apache_beam.portability import common_urns


	@functools.total_ordering
	class Timestamp(object):
	"""Represents a Unix second timestamp with microsecond granularity.

	Can be treated in common timestamp arithmetic operations as a numeric type.

	Internally stores a time interval as an int of microseconds. This strategy
	is necessary since floating point values lose precision when storing values,
	especially after arithmetic operations (for example, 10000000 % 0.1 evaluates
	to 0.0999999994448885).
	"""

	def __init__(self, seconds=0, micros=0):
	if not isinstance(seconds, (int, long, float)):
	raise TypeError('Cannot interpret %s %s as seconds.' % (
	seconds, type(seconds)))
	if not isinstance(micros, (int, long, float)):
	raise TypeError('Cannot interpret %s %s as micros.' % (
	micros, type(micros)))
	self.micros = int(seconds * 1000000) + int(micros)

	@staticmethod
	def of(seconds):
	"""Return the Timestamp for the given number of seconds.

	If the input is already a Timestamp, the input itself will be returned.

	Args:
	seconds: Number of seconds as int, float, long, or Timestamp.

	Returns:
	Corresponding Timestamp object.
	"""

	if not isinstance(seconds, (int, long, float, Timestamp)):
	raise TypeError('Cannot interpret %s %s as Timestamp.' % (
	seconds, type(seconds)))
	if isinstance(seconds, Timestamp):
	return seconds
	return Timestamp(seconds)

	RFC_3339_RE = re.compile(
	r'^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.(\d+))?Z$')

	@staticmethod
	def _epoch_datetime_utc():
	return datetime.datetime.fromtimestamp(0, pytz.utc)

	@classmethod
	def from_utc_datetime(cls, dt):
	"""Create a ``Timestamp`` instance from a ``datetime.datetime`` object.

	Args:
	dt: A ``datetime.datetime`` object in UTC (offset-aware).
	"""
	if dt.tzinfo != pytz.utc:
	raise ValueError('dt not in UTC: %s' % dt)
	duration = dt - cls._epoch_datetime_utc()
	return Timestamp(duration.total_seconds())

	@classmethod
	def from_rfc3339(cls, rfc3339):
	"""Create a ``Timestamp`` instance from an RFC 3339 compliant string.

	Args:
	rfc3339: String in RFC 3339 form.
	"""
	dt_args = []
	match = cls.RFC_3339_RE.match(rfc3339)
	if match is None:
	raise ValueError('Could not parse RFC 3339 string: %s' % rfc3339)
	for s in match.groups():
	if s is not None:
	dt_args.append(int(s))
	else:
	dt_args.append(0)
	dt_args += (pytz.utc, )
	dt = datetime.datetime(*dt_args)
	return cls.from_utc_datetime(dt)

	def predecessor(self):
	"""Returns the largest timestamp smaller than self."""
	return Timestamp(micros=self.micros - 1)

	def __repr__(self):
	micros = self.micros
	sign = ''
	if micros < 0:
	sign = '-'
	micros = -micros
	int_part = micros // 1000000
	frac_part = micros % 1000000
	if frac_part:
	return 'Timestamp(%s%d.%06d)' % (sign, int_part, frac_part)
	return 'Timestamp(%s%d)' % (sign, int_part)

	def to_utc_datetime(self):
	# We can't easily construct a datetime object from microseconds, so we
	# create one at the epoch and add an appropriate timedelta interval.
	return self._epoch_datetime_utc().replace(tzinfo=None) + datetime.timedelta(
	microseconds=self.micros)

	def to_rfc3339(self):
	# Append 'Z' for UTC timezone.
	return self.to_utc_datetime().isoformat() + 'Z'

	def __float__(self):
	# Note that the returned value may have lost precision.
	return self.micros / 1000000

	def __int__(self):
	# Note that the returned value may have lost precision.
	return self.micros // 1000000

	def __eq__(self, other):
	# Allow comparisons between Duration and Timestamp values.
	if not isinstance(other, Duration):
	other = Timestamp.of(other)
	return self.micros == other.micros

	def __ne__(self, other):
	# TODO(BEAM-5949): Needed for Python 2 compatibility.
	return not self == other

	def __lt__(self, other):
	# Allow comparisons between Duration and Timestamp values.
	if not isinstance(other, Duration):
	other = Timestamp.of(other)
	return self.micros < other.micros

	def __hash__(self):
	return hash(self.micros)

	def __add__(self, other):
	other = Duration.of(other)
	return Timestamp(micros=self.micros + other.micros)

	def __radd__(self, other):
	return self + other

	def __sub__(self, other):
	other = Duration.of(other)
	return Timestamp(micros=self.micros - other.micros)

	def __mod__(self, other):
	other = Duration.of(other)
	return Duration(micros=self.micros % other.micros)


	MIN_TIMESTAMP = Timestamp(micros=int(
	common_urns.constants.MIN_TIMESTAMP_MILLIS.constant)*1000)
	MAX_TIMESTAMP = Timestamp(micros=int(
	common_urns.constants.MAX_TIMESTAMP_MILLIS.constant)*1000)


	@functools.total_ordering
	class Duration(object):
	"""Represents a second duration with microsecond granularity.

	Can be treated in common arithmetic operations as a numeric type.

	Internally stores a time interval as an int of microseconds. This strategy
	is necessary since floating point values lose precision when storing values,
	especially after arithmetic operations (for example, 10000000 % 0.1 evaluates
	to 0.0999999994448885).
	"""

	def __init__(self, seconds=0, micros=0):
	self.micros = int(seconds * 1000000) + int(micros)

	@staticmethod
	def of(seconds):
	"""Return the Duration for the given number of seconds since Unix epoch.

	If the input is already a Duration, the input itself will be returned.

	Args:
	seconds: Number of seconds as int, float or Duration.

	Returns:
	Corresponding Duration object.
	"""

	if isinstance(seconds, Timestamp):
	raise TypeError('Cannot interpret %s as Duration.' % seconds)
	if isinstance(seconds, Duration):
	return seconds
	return Duration(seconds)

	def __repr__(self):
	micros = self.micros
	sign = ''
	if micros < 0:
	sign = '-'
	micros = -micros
	int_part = micros // 1000000
	frac_part = micros % 1000000
	if frac_part:
	return 'Duration(%s%d.%06d)' % (sign, int_part, frac_part)
	return 'Duration(%s%d)' % (sign, int_part)

	def __float__(self):
	# Note that the returned value may have lost precision.
	return self.micros / 1000000

	def __eq__(self, other):
	# Allow comparisons between Duration and Timestamp values.
	if not isinstance(other, Timestamp):
	other = Duration.of(other)
	return self.micros == other.micros

	def __ne__(self, other):
	# TODO(BEAM-5949): Needed for Python 2 compatibility.
	return not self == other

	def __lt__(self, other):
	# Allow comparisons between Duration and Timestamp values.
	if not isinstance(other, Timestamp):
	other = Duration.of(other)
	return self.micros < other.micros

	def __hash__(self):
	return hash(self.micros)

	def __neg__(self):
	return Duration(micros=-self.micros)

	def __add__(self, other):
	if isinstance(other, Timestamp):
	return other + self
	other = Duration.of(other)
	return Duration(micros=self.micros + other.micros)

	def __radd__(self, other):
	return self + other

	def __sub__(self, other):
	other = Duration.of(other)
	return Duration(micros=self.micros - other.micros)

	def __rsub__(self, other):
	return -(self - other)

	def __mul__(self, other):
	other = Duration.of(other)
	return Duration(micros=self.micros * other.micros // 1000000)

	def __rmul__(self, other):
	return self * other

	def __mod__(self, other):
	other = Duration.of(other)
	return Duration(micros=self.micros % other.micros)


	# The minimum granularity / interval expressible in a Timestamp / Duration
	# object.
	TIME_GRANULARITY = Duration(micros=1)