Skip to content

Instantly share code, notes, and snippets.

@ntim
Created February 21, 2016 13:46
Show Gist options
  • Select an option

  • Save ntim/c13b5b9bec2cbdd2da2c to your computer and use it in GitHub Desktop.

Select an option

Save ntim/c13b5b9bec2cbdd2da2c to your computer and use it in GitHub Desktop.
Luigi RangeBase implementation replacing datetime with an integer
import luigi
import datetime
import itertools
import time
class RangeInteger(luigi.tools.range.RangeBase):
"""
Instead of datetime with a specific resolution, we use a simple integer.
The rangeBase is a bit inflexible regarding the parameter type, but we can
work around it by parsing it to unix timestamps back and forth.
"""
start = luigi.Parameter(default=1)
stop = luigi.Parameter(default=1000)
# Task limit usually set to stop, to enable scheduling of all parameters.
task_limit = luigi.IntParameter(1000)
def moving_start(self, now):
return self.parameter_to_datetime(int(self.start))
def moving_stop(self, now):
return self.parameter_to_datetime(int(self.stop))
def datetime_to_parameter(self, dt):
return time.mktime(dt.timetuple())
def parameter_to_datetime(self, p):
return datetime.datetime.fromtimestamp(int(p))
def finite_datetimes(self, finite_start, finite_stop):
date_start = datetime.datetime(finite_start.year, finite_start.month, finite_start.day)
dates = []
for i in itertools.count():
t = date_start + datetime.timedelta(seconds=i)
if t >= finite_stop:
return dates
if t >= finite_start:
dates.append(t)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment