Using floating-point arithmetic for time interval calculations can lead to precision errors in scheduling. Consider using integer arithmetic with a smaller time unit (milliseconds) to avoid floating-point precision issues.
Suggested change (from a copilot review suggestion)
red block
return lcm(*[floor(i / minimum_polling_interval) for i in intervals]) * minimum_polling_interval
@staticmethod
def _separate_coprimes(intervals):
# TODO: The math in _separate_coprimes and calculate_hyperperiod needs to be done with integers, not floats.
# How to handle sub-second intervals? Should we do this using milliseconds?
separated = []
unseparated = list(intervals)
end red block
green block
# Convert intervals and minimum_polling_interval to integer milliseconds
intervals_ms = [int(i * 1000) for i in intervals]
min_polling_interval_ms = int(minimum_polling_interval * 1000)
# Calculate the number of minimum polling intervals in each interval
intervals_counts = [i // min_polling_interval_ms for i in intervals_ms]
hyperperiod_count = lcm(*intervals_counts)
return timedelta(milliseconds=hyperperiod_count * min_polling_interval_ms)
@staticmethod
def _separate_coprimes(intervals):
# Convert intervals to integer milliseconds
intervals_ms = [int(i * 1000) for i in intervals]
separated = []
unseparated = list(intervals_ms)
end green block
Using floating-point arithmetic for time interval calculations can lead to precision errors in scheduling. Consider using integer arithmetic with a smaller time unit (milliseconds) to avoid floating-point precision issues.
Suggested change (from a copilot review suggestion)
red block
return lcm(*[floor(i / minimum_polling_interval) for i in intervals]) * minimum_polling_interval
@staticmethod
def _separate_coprimes(intervals):
# TODO: The math in _separate_coprimes and calculate_hyperperiod needs to be done with integers, not floats.
# How to handle sub-second intervals? Should we do this using milliseconds?
separated = []
unseparated = list(intervals)
end red block
green block
# Convert intervals and minimum_polling_interval to integer milliseconds
intervals_ms = [int(i * 1000) for i in intervals]
min_polling_interval_ms = int(minimum_polling_interval * 1000)
# Calculate the number of minimum polling intervals in each interval
intervals_counts = [i // min_polling_interval_ms for i in intervals_ms]
hyperperiod_count = lcm(*intervals_counts)
return timedelta(milliseconds=hyperperiod_count * min_polling_interval_ms)
@staticmethod
def _separate_coprimes(intervals):
# Convert intervals to integer milliseconds
intervals_ms = [int(i * 1000) for i in intervals]
separated = []
unseparated = list(intervals_ms)
end green block