def calculate_schedule_usage_for_period()

in source/lambda/configuration/config_admin.py [0:0]

• 66 lines of code
• 19 McCabe index (conditional complexity)

    def calculate_schedule_usage_for_period(self, schedule_name, start_dt, stop_dt=None, logger=None):

        result = {}

        def running_seconds(startdt, stopdt):
            return max(int((stopdt - startdt).total_seconds()), 60)

        def running_hours(startdt, stopdt):
            return int(((stopdt - startdt).total_seconds() - 1) / 3600) + 1

        def make_period(started_dt, stopped_dt):
            running_period = ({
                "begin": started_dt,
                "end": stopped_dt,
                "billing_hours": running_hours(started_dt, stopped_dt),
                "billing_seconds": running_seconds(started_dt, stopped_dt)
            })
            return running_period

        self._logger = logger

        stop = stop_dt or start_dt
        if start_dt > stop:
            raise ValueError(ERR_STOP_MUST_BE_LATER_OR_EQUAL_TO_START)

        dt = start_dt if isinstance(start_dt, datetime) else datetime(start_dt.year, start_dt.month, start_dt.day)

        config_data = ConfigDynamodbAdapter(self._table.name).config

        while dt <= stop:

            self._configuration = SchedulerConfigBuilder(logger=self._logger).build(config_data)
            conf = configuration.SchedulerConfigBuilder(self._logger).build(config=config_data, dt=dt)
            schedule = conf.get_schedule(schedule_name)

            timeline = {dt.replace(hour=0, minute=0)}
            for p in schedule.periods:
                begintime = p["period"].begintime
                endtime = p["period"].endtime
                if begintime is None and endtime is None:
                    timeline.add(dt.replace(hour=0, minute=0))
                    timeline.add(dt.replace(hour=23, minute=59))
                else:
                    if begintime:
                        timeline.add(dt.replace(hour=begintime.hour, minute=begintime.minute))
                    if endtime:
                        timeline.add(dt.replace(hour=endtime.hour, minute=endtime.minute))

            running_periods = {}
            started = None
            starting_period = None
            current_state = None
            inst = as_namedtuple("Instance", {"instance_str": "instance", "allow_resize": False})
            for tm in sorted(list(timeline)):
                desired_state, instance_type, period = schedule.get_desired_state(inst, self._logger, tm, False)

                if current_state != desired_state:
                    if desired_state == InstanceSchedule.STATE_RUNNING:
                        started = tm
                        current_state = InstanceSchedule.STATE_RUNNING
                        starting_period = period
                    elif desired_state == InstanceSchedule.STATE_STOPPED:
                        stopped = tm
                        desired_state_with_adj_check, _, __ = schedule.get_desired_state(inst, self._logger, tm, True)
                        if desired_state_with_adj_check == InstanceSchedule.STATE_RUNNING:
                            stopped += timedelta(minutes=1)
                        if current_state == InstanceSchedule.STATE_RUNNING:
                            current_state = InstanceSchedule.STATE_STOPPED
                            running_periods[starting_period] = (make_period(started, stopped))

            if current_state == InstanceSchedule.STATE_RUNNING:
                stopped = dt.replace(hour=23, minute=59) + timedelta(minutes=1)
                running_periods[starting_period] = (make_period(started, stopped))

            result[str(dt.date())] = {
                "running_periods": running_periods,
                "billing_seconds": sum([running_periods[ps]["billing_seconds"] for ps in running_periods]),
                "billing_hours": sum([running_periods[ph]["billing_hours"] for ph in running_periods])
            }

            dt += timedelta(days=1)

        return {"schedule": schedule_name, "usage": result}