1
0
Fork 0
Browse Source

implement find_flicker feature.

This functions goes through the list of readings and forms a
dictionary of time-stamps, with light readings beyond a specified
readings-per-second threshold, and the reading for said time-stamp.

The results are then saved to the specified file, using
'save_rps_totals' function.
stable
Craig Oates 3 years ago
parent
commit
7bfd06cf03
  1. 14
      src/flicker.py
  2. 22
      src/services/data_services.py

14
src/flicker.py

@ -4,21 +4,23 @@ from services import io_services, log_services, data_services
# import pdb # import pdb
def main(): def main():
raw_data_path = "data/test-data-lite.csv" raw_data_path = "data/test-data.csv"
rps_save_path = "data/results/readings-per-sec.csv" rps_save_path = "data/results/readings-per-sec.csv"
rps_above_thresh = "data/results/readings_above_threshold.csv" rps_above_thresh = "data/results/readings_above_threshold.csv"
flicker_list = "data/results/flicker_list.csv"
raw_data = io_services.load_raw_data(raw_data_path) raw_data = io_services.load_raw_data(raw_data_path)
# log_services.print_list(raw_data) # log_services.print_list(raw_data)
time_tallies = data_services.tally_readings_per_second(raw_data) time_tallies = data_services.tally_readings_per_second(raw_data)
# log_services.print_dictionary(time_tallies) # log_services.print_dictionary(time_tallies)
rps_totals = data_services.total_count_for_each_reading_per_second(time_tallies) rps_totals = data_services.total_count_for_each_reading_per_second(time_tallies)
# log_services.print_dictionary(rps_totals) # log_services.print_dictionary(rps_totals)
# io_services.save_rps_totals(rps_totals, rps_save_path) io_services.save_rps_totals(rps_totals, rps_save_path)
rps_above_two = data_services.get_rps_above(2, time_tallies) rps_above_herz = data_services.get_rps_above(4, time_tallies)
# log_services.print_list(rps_above_two) # log_services.print_list(rps_above_two)
io_services.save_rps_above_threshold(rps_above_two, rps_above_thresh) io_services.save_rps_above_threshold(rps_above_herz, rps_above_thresh)
# flicker_tallies = data_services.tally_flickers(rsp_above_two, raw_data) flicker_entries = data_services.find_flickers(rps_above_herz, raw_data)
# log_services.print_dictionary(flicker_tallies) # log_services.print_dictionary(flicker_tallies)
io_services.save_rps_totals(flicker_tallies, flicker_entries)
# filtered_flicker_entries =
if __name__ == "__main__": if __name__ == "__main__":
main() main()

22
src/services/data_services.py

@ -28,13 +28,19 @@ def get_rps_above(threshold, readings):
times.append(k) times.append(k)
return times return times
def tally_flickers(selected_readings, all_readings): def find_flickers(selected_readings, all_readings):
selected = dict() selected = dict()
log_services.print_list(selected_readings) # log_services.print_list(selected_readings)
# for time in selected_readings: # print(all_readings[1])
# print(type(all_readings[0]))
# for time in selected_readings: for current_selected_reading in selected_readings:
# # print(f"Looking for {time}...") readings = list()
# selected[time] = all_readings[time] for current_reading in all_readings:
# # for k, v in all_readings.items(): # print(f"Checking {item} is in {x}...")
if (current_selected_reading == current_reading[0]):
# selected[item] = x[1]
# print(f"Added {item}.")
# selected.append( (item, x[1]) )
readings.append(current_reading[1])
selected[current_selected_reading] = readings
return selected return selected