1
0
Fork 0
Browse Source

filter readings (over herz and light levels threshold).

Having filtered down the list to readings which suppass the Hertz
threshold (4+ per-second at time of writing), the code here filters it
down even more. This bit of code searches for readings within this
already filtered list for any readings which activate the light in the
gallery (with the threshold matching that of 'gallery1' which in
anything over 39). It then proceeds to save the results.
stable
Craig Oates 3 years ago
parent
commit
2e602e9082
  1. 13
      src/flicker.py
  2. 20
      src/services/data_services.py

13
src/flicker.py

@ -1,13 +1,13 @@
import csv
import datetime
from services import io_services, log_services, data_services
# import pdb
def main():
# File paths used to save and load the data.
raw_data_path = "data/test-data.csv"
rps_save_path = "data/results/readings-per-sec.csv"
rps_above_thresh = "data/results/readings_above_threshold.csv"
flicker_list = "data/results/flicker_list.csv"
filtered_flickers = "data/results/filtered_flicker_entries.csv"
raw_data = io_services.load_raw_data(raw_data_path)
# log_services.print_list(raw_data)
time_tallies = data_services.tally_readings_per_second(raw_data)
@ -20,7 +20,10 @@ def main():
io_services.save_rps_above_threshold(rps_above_herz, rps_above_thresh)
flicker_entries = data_services.find_flickers(rps_above_herz, raw_data)
# log_services.print_dictionary(flicker_tallies)
io_services.save_rps_totals(flicker_tallies, flicker_entries)
# filtered_flicker_entries =
io_services.save_rps_totals(flicker_entries, flicker_list)
filtered_flicker_entries = data_services.find_readings_with_lights_on(flicker_entries)
# log_services.print_list(filtered_flicker_entries)
io_services.save_filtered_flickers(filtered_flicker_entries, filtered_flickers)
if __name__ == "__main__":
main()

20
src/services/data_services.py

@ -1,5 +1,3 @@
from services import log_services
def tally_readings_per_second(data):
time_tallies = dict()
for item in data:
@ -21,26 +19,26 @@ def total_count_for_each_reading_per_second(time_tallies):
def get_rps_above(threshold, readings):
times = list()
# print(readings.values())
for k, v in readings.items():
# print([k, v])
if (v >= threshold):
times.append(k)
return times
def find_flickers(selected_readings, all_readings):
selected = dict()
# log_services.print_list(selected_readings)
# print(all_readings[1])
# print(type(all_readings[0]))
for current_selected_reading in selected_readings:
readings = list()
for current_reading in all_readings:
# 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
def find_readings_with_lights_on(flicker_entries):
filtered_readings = list()
for time, reading in flicker_entries.items():
for item in reading:
val = int(item)
if (val > 39):
filtered_readings.append( (time, reading) )
return filtered_readings