You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ro...@apache.org on 2016/11/01 05:14:45 UTC
[1/3] incubator-beam git commit: Move Timestamp and related classes
into apache_beam/utils.
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk 979e299c7 -> 13dd9ff49
Move Timestamp and related classes into apache_beam/utils.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8fd651f0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8fd651f0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8fd651f0
Branch: refs/heads/python-sdk
Commit: 8fd651f05d1e769000fa093e6ab626ef2ac87dec
Parents: 979e299
Author: Robert Bradshaw <ro...@gmail.com>
Authored: Mon Oct 31 16:47:24 2016 -0700
Committer: Robert Bradshaw <ro...@gmail.com>
Committed: Mon Oct 31 16:47:24 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/transforms/timeutil.py | 197 +----------------
.../apache_beam/transforms/timeutil_test.py | 168 --------------
sdks/python/apache_beam/utils/timestamp.py | 217 +++++++++++++++++++
sdks/python/apache_beam/utils/timestamp_test.py | 168 ++++++++++++++
sdks/python/apache_beam/utils/windowed_value.py | 6 +-
5 files changed, 394 insertions(+), 362 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8fd651f0/sdks/python/apache_beam/transforms/timeutil.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/timeutil.py b/sdks/python/apache_beam/transforms/timeutil.py
index f026d83..749cda1 100644
--- a/sdks/python/apache_beam/transforms/timeutil.py
+++ b/sdks/python/apache_beam/transforms/timeutil.py
@@ -15,7 +15,7 @@
# limitations under the License.
#
-"""Time and timer utilities."""
+"""Timestamp utilities."""
from __future__ import absolute_import
@@ -25,196 +25,11 @@ from abc import abstractmethod
import datetime
-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):
- 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 or Timestamp.
-
- Returns:
- Corresponding Timestamp object.
- """
-
- if isinstance(seconds, Duration):
- raise TypeError('Can\'t interpret %s as Timestamp.' % seconds)
- if isinstance(seconds, Timestamp):
- return seconds
- return Timestamp(seconds)
-
- 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)
- else:
- return 'Timestamp(%s%d)' % (sign, int_part)
-
- def to_utc_datetime(self):
- epoch = datetime.datetime.utcfromtimestamp(0)
- # We can't easily construct a datetime object from microseconds, so we
- # create one at the epoch and add an appropriate timedelta interval.
- return epoch + datetime.timedelta(microseconds=self.micros)
-
- def isoformat(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 float(self.micros) / 1000000
-
- def __int__(self):
- # Note that the returned value may have lost precision.
- return self.micros / 1000000
-
- def __cmp__(self, other):
- # Allow comparisons between Duration and Timestamp values.
- if not isinstance(other, Duration):
- other = Timestamp.of(other)
- return cmp(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=-0x7fffffffffffffff - 1)
-MAX_TIMESTAMP = Timestamp(micros=0x7fffffffffffffff)
-
-
-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('Can\'t 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)
- else:
- return 'Duration(%s%d)' % (sign, int_part)
-
- def __float__(self):
- # Note that the returned value may have lost precision.
- return float(self.micros) / 1000000
-
- def __int__(self):
- # Note that the returned value may have lost precision.
- return self.micros / 1000000
-
- def __cmp__(self, other):
- # Allow comparisons between Duration and Timestamp values.
- if not isinstance(other, Timestamp):
- other = Duration.of(other)
- return cmp(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)
-
+# For backwards compatibility
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import MAX_TIMESTAMP
+from apache_beam.utils.timestamp import MIN_TIMESTAMP
+from apache_beam.utils.timestamp import Timestamp
class TimeDomain(object):
"""Time domain for streaming timers."""
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8fd651f0/sdks/python/apache_beam/transforms/timeutil_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/timeutil_test.py b/sdks/python/apache_beam/transforms/timeutil_test.py
deleted file mode 100644
index a455404..0000000
--- a/sdks/python/apache_beam/transforms/timeutil_test.py
+++ /dev/null
@@ -1,168 +0,0 @@
-#
-# 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.
-#
-
-"""Unit tests for time utilities."""
-
-from __future__ import absolute_import
-
-import unittest
-
-from apache_beam.transforms.timeutil import Duration
-from apache_beam.transforms.timeutil import Timestamp
-
-
-class TimestampTest(unittest.TestCase):
-
- def test_of(self):
- interval = Timestamp(123)
- self.assertEqual(id(interval), id(Timestamp.of(interval)))
- self.assertEqual(interval, Timestamp.of(123.0))
- with self.assertRaises(TypeError):
- Timestamp.of(Duration(10))
-
- def test_precision(self):
- self.assertEqual(Timestamp(10000000) % 0.1, 0)
- self.assertEqual(Timestamp(10000000) % 0.05, 0)
- self.assertEqual(Timestamp(10000000) % 0.000005, 0)
- self.assertEqual(Timestamp(10000000) % Duration(0.1), 0)
- self.assertEqual(Timestamp(10000000) % Duration(0.05), 0)
- self.assertEqual(Timestamp(10000000) % Duration(0.000005), 0)
-
- def test_utc_timestamp(self):
- self.assertEqual(Timestamp(10000000).isoformat(),
- '1970-04-26T17:46:40Z')
- self.assertEqual(Timestamp(10000000.000001).isoformat(),
- '1970-04-26T17:46:40.000001Z')
- self.assertEqual(Timestamp(1458343379.123456).isoformat(),
- '2016-03-18T23:22:59.123456Z')
-
- def test_arithmetic(self):
- # Supported operations.
- self.assertEqual(Timestamp(123) + 456, 579)
- self.assertEqual(Timestamp(123) + Duration(456), 579)
- self.assertEqual(456 + Timestamp(123), 579)
- self.assertEqual(Duration(456) + Timestamp(123), 579)
- self.assertEqual(Timestamp(123) - 456, -333)
- self.assertEqual(Timestamp(123) - Duration(456), -333)
- self.assertEqual(Timestamp(1230) % 456, 318)
- self.assertEqual(Timestamp(1230) % Duration(456), 318)
-
- # Check that direct comparison of Timestamp and Duration is allowed.
- self.assertTrue(Duration(123) == Timestamp(123))
- self.assertTrue(Timestamp(123) == Duration(123))
- self.assertFalse(Duration(123) == Timestamp(1230))
- self.assertFalse(Timestamp(123) == Duration(1230))
-
- # Check return types.
- self.assertEqual((Timestamp(123) + 456).__class__, Timestamp)
- self.assertEqual((Timestamp(123) + Duration(456)).__class__, Timestamp)
- self.assertEqual((456 + Timestamp(123)).__class__, Timestamp)
- self.assertEqual((Duration(456) + Timestamp(123)).__class__, Timestamp)
- self.assertEqual((Timestamp(123) - 456).__class__, Timestamp)
- self.assertEqual((Timestamp(123) - Duration(456)).__class__, Timestamp)
- self.assertEqual((Timestamp(1230) % 456).__class__, Duration)
- self.assertEqual((Timestamp(1230) % Duration(456)).__class__, Duration)
-
- # Unsupported operations.
- with self.assertRaises(TypeError):
- self.assertEqual(Timestamp(123) * 456, 56088)
- with self.assertRaises(TypeError):
- self.assertEqual(Timestamp(123) * Duration(456), 56088)
- with self.assertRaises(TypeError):
- self.assertEqual(456 * Timestamp(123), 56088)
- with self.assertRaises(TypeError):
- self.assertEqual(Duration(456) * Timestamp(123), 56088)
- with self.assertRaises(TypeError):
- self.assertEqual(456 - Timestamp(123), 333)
- with self.assertRaises(TypeError):
- self.assertEqual(Duration(456) - Timestamp(123), 333)
- with self.assertRaises(TypeError):
- self.assertEqual(-Timestamp(123), -123)
- with self.assertRaises(TypeError):
- self.assertEqual(-Timestamp(123), -Duration(123))
- with self.assertRaises(TypeError):
- self.assertEqual(1230 % Timestamp(456), 318)
- with self.assertRaises(TypeError):
- self.assertEqual(Duration(1230) % Timestamp(456), 318)
-
- def test_sort_order(self):
- self.assertEqual(
- [-63, Timestamp(-3), 2, 9, Timestamp(292.3), 500],
- sorted([9, 2, Timestamp(-3), Timestamp(292.3), -63, 500]))
- self.assertEqual(
- [4, 5, Timestamp(6), Timestamp(7), 8, 9],
- sorted([9, 8, Timestamp(7), Timestamp(6), 5, 4]))
-
- def test_str(self):
- self.assertEqual('Timestamp(1.234567)',
- str(Timestamp(1.234567)))
- self.assertEqual('Timestamp(-1.234567)',
- str(Timestamp(-1.234567)))
- self.assertEqual('Timestamp(-999999999.900000)',
- str(Timestamp(-999999999.9)))
- self.assertEqual('Timestamp(999999999)',
- str(Timestamp(999999999)))
- self.assertEqual('Timestamp(-999999999)',
- str(Timestamp(-999999999)))
-
-
-class DurationTest(unittest.TestCase):
-
- def test_of(self):
- interval = Duration(123)
- self.assertEqual(id(interval), id(Duration.of(interval)))
- self.assertEqual(interval, Duration.of(123.0))
- with self.assertRaises(TypeError):
- Duration.of(Timestamp(10))
-
- def test_precision(self):
- self.assertEqual(Duration(10000000) % 0.1, 0)
- self.assertEqual(Duration(10000000) % 0.05, 0)
- self.assertEqual(Duration(10000000) % 0.000005, 0)
-
- def test_arithmetic(self):
- self.assertEqual(Duration(123) + 456, 579)
- self.assertEqual(456 + Duration(123), 579)
- self.assertEqual(Duration(123) * 456, 56088)
- self.assertEqual(456 * Duration(123), 56088)
- self.assertEqual(Duration(123) - 456, -333)
- self.assertEqual(456 - Duration(123), 333)
- self.assertEqual(-Duration(123), -123)
-
- def test_sort_order(self):
- self.assertEqual(
- [-63, Duration(-3), 2, 9, Duration(292.3), 500],
- sorted([9, 2, Duration(-3), Duration(292.3), -63, 500]))
- self.assertEqual(
- [4, 5, Duration(6), Duration(7), 8, 9],
- sorted([9, 8, Duration(7), Duration(6), 5, 4]))
-
- def test_str(self):
- self.assertEqual('Duration(1.234567)',
- str(Duration(1.234567)))
- self.assertEqual('Duration(-1.234567)',
- str(Duration(-1.234567)))
- self.assertEqual('Duration(-999999999.900000)',
- str(Duration(-999999999.9)))
- self.assertEqual('Duration(999999999)',
- str(Duration(999999999)))
- self.assertEqual('Duration(-999999999)',
- str(Duration(-999999999)))
-
-
-if __name__ == '__main__':
- unittest.main()
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8fd651f0/sdks/python/apache_beam/utils/timestamp.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/timestamp.py b/sdks/python/apache_beam/utils/timestamp.py
new file mode 100644
index 0000000..1a019c3
--- /dev/null
+++ b/sdks/python/apache_beam/utils/timestamp.py
@@ -0,0 +1,217 @@
+#
+# 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."""
+
+from __future__ import absolute_import
+
+from abc import ABCMeta
+from abc import abstractmethod
+
+import datetime
+
+
+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):
+ 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 or Timestamp.
+
+ Returns:
+ Corresponding Timestamp object.
+ """
+
+ if isinstance(seconds, Duration):
+ raise TypeError('Can\'t interpret %s as Timestamp.' % seconds)
+ if isinstance(seconds, Timestamp):
+ return seconds
+ return Timestamp(seconds)
+
+ 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)
+ else:
+ return 'Timestamp(%s%d)' % (sign, int_part)
+
+ def to_utc_datetime(self):
+ epoch = datetime.datetime.utcfromtimestamp(0)
+ # We can't easily construct a datetime object from microseconds, so we
+ # create one at the epoch and add an appropriate timedelta interval.
+ return epoch + datetime.timedelta(microseconds=self.micros)
+
+ def isoformat(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 float(self.micros) / 1000000
+
+ def __int__(self):
+ # Note that the returned value may have lost precision.
+ return self.micros / 1000000
+
+ def __cmp__(self, other):
+ # Allow comparisons between Duration and Timestamp values.
+ if not isinstance(other, Duration):
+ other = Timestamp.of(other)
+ return cmp(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=-0x7fffffffffffffff - 1)
+MAX_TIMESTAMP = Timestamp(micros=0x7fffffffffffffff)
+
+
+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('Can\'t 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)
+ else:
+ return 'Duration(%s%d)' % (sign, int_part)
+
+ def __float__(self):
+ # Note that the returned value may have lost precision.
+ return float(self.micros) / 1000000
+
+ def __int__(self):
+ # Note that the returned value may have lost precision.
+ return self.micros / 1000000
+
+ def __cmp__(self, other):
+ # Allow comparisons between Duration and Timestamp values.
+ if not isinstance(other, Timestamp):
+ other = Duration.of(other)
+ return cmp(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)
+
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8fd651f0/sdks/python/apache_beam/utils/timestamp_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/timestamp_test.py b/sdks/python/apache_beam/utils/timestamp_test.py
new file mode 100644
index 0000000..3322936
--- /dev/null
+++ b/sdks/python/apache_beam/utils/timestamp_test.py
@@ -0,0 +1,168 @@
+#
+# 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.
+#
+
+"""Unit tests for time utilities."""
+
+from __future__ import absolute_import
+
+import unittest
+
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+
+class TimestampTest(unittest.TestCase):
+
+ def test_of(self):
+ interval = Timestamp(123)
+ self.assertEqual(id(interval), id(Timestamp.of(interval)))
+ self.assertEqual(interval, Timestamp.of(123.0))
+ with self.assertRaises(TypeError):
+ Timestamp.of(Duration(10))
+
+ def test_precision(self):
+ self.assertEqual(Timestamp(10000000) % 0.1, 0)
+ self.assertEqual(Timestamp(10000000) % 0.05, 0)
+ self.assertEqual(Timestamp(10000000) % 0.000005, 0)
+ self.assertEqual(Timestamp(10000000) % Duration(0.1), 0)
+ self.assertEqual(Timestamp(10000000) % Duration(0.05), 0)
+ self.assertEqual(Timestamp(10000000) % Duration(0.000005), 0)
+
+ def test_utc_timestamp(self):
+ self.assertEqual(Timestamp(10000000).isoformat(),
+ '1970-04-26T17:46:40Z')
+ self.assertEqual(Timestamp(10000000.000001).isoformat(),
+ '1970-04-26T17:46:40.000001Z')
+ self.assertEqual(Timestamp(1458343379.123456).isoformat(),
+ '2016-03-18T23:22:59.123456Z')
+
+ def test_arithmetic(self):
+ # Supported operations.
+ self.assertEqual(Timestamp(123) + 456, 579)
+ self.assertEqual(Timestamp(123) + Duration(456), 579)
+ self.assertEqual(456 + Timestamp(123), 579)
+ self.assertEqual(Duration(456) + Timestamp(123), 579)
+ self.assertEqual(Timestamp(123) - 456, -333)
+ self.assertEqual(Timestamp(123) - Duration(456), -333)
+ self.assertEqual(Timestamp(1230) % 456, 318)
+ self.assertEqual(Timestamp(1230) % Duration(456), 318)
+
+ # Check that direct comparison of Timestamp and Duration is allowed.
+ self.assertTrue(Duration(123) == Timestamp(123))
+ self.assertTrue(Timestamp(123) == Duration(123))
+ self.assertFalse(Duration(123) == Timestamp(1230))
+ self.assertFalse(Timestamp(123) == Duration(1230))
+
+ # Check return types.
+ self.assertEqual((Timestamp(123) + 456).__class__, Timestamp)
+ self.assertEqual((Timestamp(123) + Duration(456)).__class__, Timestamp)
+ self.assertEqual((456 + Timestamp(123)).__class__, Timestamp)
+ self.assertEqual((Duration(456) + Timestamp(123)).__class__, Timestamp)
+ self.assertEqual((Timestamp(123) - 456).__class__, Timestamp)
+ self.assertEqual((Timestamp(123) - Duration(456)).__class__, Timestamp)
+ self.assertEqual((Timestamp(1230) % 456).__class__, Duration)
+ self.assertEqual((Timestamp(1230) % Duration(456)).__class__, Duration)
+
+ # Unsupported operations.
+ with self.assertRaises(TypeError):
+ self.assertEqual(Timestamp(123) * 456, 56088)
+ with self.assertRaises(TypeError):
+ self.assertEqual(Timestamp(123) * Duration(456), 56088)
+ with self.assertRaises(TypeError):
+ self.assertEqual(456 * Timestamp(123), 56088)
+ with self.assertRaises(TypeError):
+ self.assertEqual(Duration(456) * Timestamp(123), 56088)
+ with self.assertRaises(TypeError):
+ self.assertEqual(456 - Timestamp(123), 333)
+ with self.assertRaises(TypeError):
+ self.assertEqual(Duration(456) - Timestamp(123), 333)
+ with self.assertRaises(TypeError):
+ self.assertEqual(-Timestamp(123), -123)
+ with self.assertRaises(TypeError):
+ self.assertEqual(-Timestamp(123), -Duration(123))
+ with self.assertRaises(TypeError):
+ self.assertEqual(1230 % Timestamp(456), 318)
+ with self.assertRaises(TypeError):
+ self.assertEqual(Duration(1230) % Timestamp(456), 318)
+
+ def test_sort_order(self):
+ self.assertEqual(
+ [-63, Timestamp(-3), 2, 9, Timestamp(292.3), 500],
+ sorted([9, 2, Timestamp(-3), Timestamp(292.3), -63, 500]))
+ self.assertEqual(
+ [4, 5, Timestamp(6), Timestamp(7), 8, 9],
+ sorted([9, 8, Timestamp(7), Timestamp(6), 5, 4]))
+
+ def test_str(self):
+ self.assertEqual('Timestamp(1.234567)',
+ str(Timestamp(1.234567)))
+ self.assertEqual('Timestamp(-1.234567)',
+ str(Timestamp(-1.234567)))
+ self.assertEqual('Timestamp(-999999999.900000)',
+ str(Timestamp(-999999999.9)))
+ self.assertEqual('Timestamp(999999999)',
+ str(Timestamp(999999999)))
+ self.assertEqual('Timestamp(-999999999)',
+ str(Timestamp(-999999999)))
+
+
+class DurationTest(unittest.TestCase):
+
+ def test_of(self):
+ interval = Duration(123)
+ self.assertEqual(id(interval), id(Duration.of(interval)))
+ self.assertEqual(interval, Duration.of(123.0))
+ with self.assertRaises(TypeError):
+ Duration.of(Timestamp(10))
+
+ def test_precision(self):
+ self.assertEqual(Duration(10000000) % 0.1, 0)
+ self.assertEqual(Duration(10000000) % 0.05, 0)
+ self.assertEqual(Duration(10000000) % 0.000005, 0)
+
+ def test_arithmetic(self):
+ self.assertEqual(Duration(123) + 456, 579)
+ self.assertEqual(456 + Duration(123), 579)
+ self.assertEqual(Duration(123) * 456, 56088)
+ self.assertEqual(456 * Duration(123), 56088)
+ self.assertEqual(Duration(123) - 456, -333)
+ self.assertEqual(456 - Duration(123), 333)
+ self.assertEqual(-Duration(123), -123)
+
+ def test_sort_order(self):
+ self.assertEqual(
+ [-63, Duration(-3), 2, 9, Duration(292.3), 500],
+ sorted([9, 2, Duration(-3), Duration(292.3), -63, 500]))
+ self.assertEqual(
+ [4, 5, Duration(6), Duration(7), 8, 9],
+ sorted([9, 8, Duration(7), Duration(6), 5, 4]))
+
+ def test_str(self):
+ self.assertEqual('Duration(1.234567)',
+ str(Duration(1.234567)))
+ self.assertEqual('Duration(-1.234567)',
+ str(Duration(-1.234567)))
+ self.assertEqual('Duration(-999999999.900000)',
+ str(Duration(-999999999.9)))
+ self.assertEqual('Duration(999999999)',
+ str(Duration(999999999)))
+ self.assertEqual('Duration(-999999999)',
+ str(Duration(-999999999)))
+
+
+if __name__ == '__main__':
+ unittest.main()
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8fd651f0/sdks/python/apache_beam/utils/windowed_value.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/windowed_value.py b/sdks/python/apache_beam/utils/windowed_value.py
index 4c50e72..35cc52f 100644
--- a/sdks/python/apache_beam/utils/windowed_value.py
+++ b/sdks/python/apache_beam/utils/windowed_value.py
@@ -25,9 +25,9 @@
#cython: profile=True
-from apache_beam.transforms.timeutil import MAX_TIMESTAMP
-from apache_beam.transforms.timeutil import MIN_TIMESTAMP
-from apache_beam.transforms.timeutil import Timestamp
+from apache_beam.utils.timestamp import MAX_TIMESTAMP
+from apache_beam.utils.timestamp import MIN_TIMESTAMP
+from apache_beam.utils.timestamp import Timestamp
class WindowedValue(object):
[3/3] incubator-beam git commit: Closes #1241
Posted by ro...@apache.org.
Closes #1241
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/13dd9ff4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/13dd9ff4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/13dd9ff4
Branch: refs/heads/python-sdk
Commit: 13dd9ff497339be86b5956eeed083cd057c6ea02
Parents: 979e299 066b7c6
Author: Robert Bradshaw <ro...@gmail.com>
Authored: Mon Oct 31 22:14:18 2016 -0700
Committer: Robert Bradshaw <ro...@gmail.com>
Committed: Mon Oct 31 22:14:18 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/coders/coder_impl.py | 17 +-
sdks/python/apache_beam/coders/coders.py | 5 +-
.../apache_beam/coders/coders_test_common.py | 11 +-
sdks/python/apache_beam/transforms/timeutil.py | 201 +----------------
.../apache_beam/transforms/timeutil_test.py | 168 ---------------
sdks/python/apache_beam/utils/timestamp.py | 213 +++++++++++++++++++
sdks/python/apache_beam/utils/timestamp_test.py | 168 +++++++++++++++
sdks/python/apache_beam/utils/windowed_value.py | 6 +-
8 files changed, 403 insertions(+), 386 deletions(-)
----------------------------------------------------------------------
[2/3] incubator-beam git commit: Remove fake Timestamp and
WindowedValue hacks.
Posted by ro...@apache.org.
Remove fake Timestamp and WindowedValue hacks.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/066b7c68
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/066b7c68
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/066b7c68
Branch: refs/heads/python-sdk
Commit: 066b7c684d876666e47301271771dca69662945b
Parents: 8fd651f
Author: Robert Bradshaw <ro...@gmail.com>
Authored: Mon Oct 31 17:26:52 2016 -0700
Committer: Robert Bradshaw <ro...@gmail.com>
Committed: Mon Oct 31 22:14:18 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/coders/coder_impl.py | 17 +++--------------
sdks/python/apache_beam/coders/coders.py | 5 +----
.../apache_beam/coders/coders_test_common.py | 11 ++++++-----
sdks/python/apache_beam/transforms/timeutil.py | 8 +++++---
sdks/python/apache_beam/utils/timestamp.py | 4 ----
5 files changed, 15 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/066b7c68/sdks/python/apache_beam/coders/coder_impl.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index c73dd31..1ff3a44 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -25,22 +25,14 @@ This module may be optionally compiled with Cython, using the corresponding
coder_impl.pxd file for type hints.
"""
-import collections
from types import NoneType
from apache_beam.coders import observable
-
+from apache_beam.utils.timestamp import Timestamp
+from apache_beam.utils.windowed_value import WindowedValue
# pylint: disable=wrong-import-order, wrong-import-position, ungrouped-imports
try:
- # Don't depend on the full dataflow sdk to test coders.
- from apache_beam.transforms.window import WindowedValue
-except ImportError:
- WindowedValue = collections.namedtuple(
- 'WindowedValue', ('value', 'timestamp', 'windows'))
-
-
-try:
from stream import InputStream as create_InputStream
from stream import OutputStream as create_OutputStream
from stream import ByteCountingOutputStream
@@ -326,14 +318,11 @@ class FloatCoderImpl(StreamCoderImpl):
class TimestampCoderImpl(StreamCoderImpl):
- def __init__(self, timestamp_class):
- self.timestamp_class = timestamp_class
-
def encode_to_stream(self, value, out, nested):
out.write_bigendian_int64(value.micros)
def decode_from_stream(self, in_stream, nested):
- return self.timestamp_class(micros=in_stream.read_bigendian_int64())
+ return Timestamp(micros=in_stream.read_bigendian_int64())
def estimate_size(self, unused_value, nested=False):
# A Timestamp is encoded as a 64-bit integer in 8 bytes, regardless of
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/066b7c68/sdks/python/apache_beam/coders/coders.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py
index da06db1..fbbc325 100644
--- a/sdks/python/apache_beam/coders/coders.py
+++ b/sdks/python/apache_beam/coders/coders.py
@@ -18,7 +18,6 @@
"""Collection of useful coders."""
import base64
-import collections
import cPickle as pickle
from apache_beam.coders import coder_impl
@@ -30,12 +29,10 @@ try:
# Import dill from the pickler module to make sure our monkey-patching of dill
# occurs.
from apache_beam.internal.pickler import dill
- from apache_beam.transforms.timeutil import Timestamp
except ImportError:
# We fall back to using the stock dill library in tests that don't use the
# full Python SDK.
import dill
- Timestamp = collections.namedtuple('Timestamp', 'micros')
def serialize_coder(coder):
@@ -265,7 +262,7 @@ class TimestampCoder(FastCoder):
"""A coder used for timeutil.Timestamp values."""
def _create_impl(self):
- return coder_impl.TimestampCoderImpl(Timestamp)
+ return coder_impl.TimestampCoderImpl()
def is_deterministic(self):
return True
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/066b7c68/sdks/python/apache_beam/coders/coders_test_common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coders_test_common.py b/sdks/python/apache_beam/coders/coders_test_common.py
index a906437..e7780e4 100644
--- a/sdks/python/apache_beam/coders/coders_test_common.py
+++ b/sdks/python/apache_beam/coders/coders_test_common.py
@@ -25,6 +25,7 @@ import dill
import coders
import observable
+from apache_beam.utils import timestamp
# Defined out of line for picklability.
@@ -160,13 +161,13 @@ class CodersTest(unittest.TestCase):
def test_timestamp_coder(self):
self.check_coder(coders.TimestampCoder(),
- *[coders.Timestamp(micros=x) for x in range(-100, 100)])
+ *[timestamp.Timestamp(micros=x) for x in range(-100, 100)])
self.check_coder(coders.TimestampCoder(),
- coders.Timestamp(micros=-1234567890),
- coders.Timestamp(micros=1234567890))
+ timestamp.Timestamp(micros=-1234567890),
+ timestamp.Timestamp(micros=1234567890))
self.check_coder(coders.TimestampCoder(),
- coders.Timestamp(micros=-1234567890123456789),
- coders.Timestamp(micros=1234567890123456789))
+ timestamp.Timestamp(micros=-1234567890123456789),
+ timestamp.Timestamp(micros=1234567890123456789))
def test_tuple_coder(self):
self.check_coder(
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/066b7c68/sdks/python/apache_beam/transforms/timeutil.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/timeutil.py b/sdks/python/apache_beam/transforms/timeutil.py
index 749cda1..5453b20 100644
--- a/sdks/python/apache_beam/transforms/timeutil.py
+++ b/sdks/python/apache_beam/transforms/timeutil.py
@@ -22,14 +22,16 @@ from __future__ import absolute_import
from abc import ABCMeta
from abc import abstractmethod
-import datetime
-
-# For backwards compatibility
+# For backwards compatibility.
+# TODO(robertwb): Remove.
+# pylint: disable=unused-import
from apache_beam.utils.timestamp import Duration
from apache_beam.utils.timestamp import MAX_TIMESTAMP
from apache_beam.utils.timestamp import MIN_TIMESTAMP
from apache_beam.utils.timestamp import Timestamp
+# pylint: enable=unused-import
+
class TimeDomain(object):
"""Time domain for streaming timers."""
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/066b7c68/sdks/python/apache_beam/utils/timestamp.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/timestamp.py b/sdks/python/apache_beam/utils/timestamp.py
index 1a019c3..cfabb77 100644
--- a/sdks/python/apache_beam/utils/timestamp.py
+++ b/sdks/python/apache_beam/utils/timestamp.py
@@ -19,9 +19,6 @@
from __future__ import absolute_import
-from abc import ABCMeta
-from abc import abstractmethod
-
import datetime
@@ -214,4 +211,3 @@ class Duration(object):
def __mod__(self, other):
other = Duration.of(other)
return Duration(micros=self.micros % other.micros)
-