- from urllib.request import *
- from json import *
- from datetime import *
- from operator import *
- import math
- class Earthquake:
- def __init__(self, place, mag, longitude, latitude, time):
- # place - string
- # mag - float
- # longitude - float
- # latitude - float
- # time - int
- self.place = place
- self.mag = mag
- self.longitude = longitude
- self.latitude = latitude
- self.time = time
- def __eq__(self, other):
- return self.place == other.place and math.isclose(self.mag, other.mag) and math.isclose(self.longitude, other.longitude) and math.isclose(self.latitude, other.latitude) and self.time == other.time
- # Print object function. Print each object with correct formatting #
- # Move the printEarthquake down because of function order #
- def getChoice():
- print("Options:")
- print(" (s)ort")
- print(" (f)ilter")
- print(" (n)ew quakes")
- print(" (q)uit")
- print("")
- choice = input("Choice: ").lower()
- return choice
- def quake_from_feature(f):
- place = f["properties"]["place"]
- mag = f["properties"]["mag"]
- longitude = f["geometry"]["coordinates"][0]
- latitude = f["geometry"]["coordinates"][1]
- badTime = f["properties"]["time"]
- time = badTime / 1000
- earthquake = Earthquake(place, float(mag), float(longitude), float(latitude), int(time))
- return earthquake
- def read_quakes_from_file(filename):
- inFile = open(filename, "r")
- newlist = []
- for line in inFile:
- linelist = line.split()
- place = " ".join(linelist[4:])
- mag = linelist[0]
- longitude = linelist[1]
- latitude = linelist[2]
- time = linelist[3]
- earthquake = Earthquake(place, float(mag), float(longitude), float(latitude), int(time))
- newlist.append(earthquake)
- inFile.close()
- return newlist
- def filter_by_mag(quakes, low, high):
- newlist = []
- for x in quakes:
- if (x.mag >= low) and (x.mag <= high):
- newlist.append(x)
- return newlist
- def filter_by_place(quakes, word):
- newlist = []
- for x in quakes:
- status = x.place.lower().find(word.lower())
- if status > 0:
- newlist.append(x)
- return newlist
- def get_json(url):
- with urlopen(url) as response:
- html = response.read()
- htmlstr = html.decode("utf-8")
- return loads(htmlstr)
- def time_to_str(time):
- return datetime.fromtimestamp(time).strftime('%Y-%m-%d %H:%M:%S')
- def printEarthquake(earthquakes):
- print("Earthquakes:")
- print("------------")
- for x in earthquakes:
- time = time_to_str(x.time)
- print("(%.2f" % x.mag + ") %40s" % x.place, "at", time, "(%7.3f, %.3f)" % (x.longitude, x.latitude))
- print("")
- def choice_S(allQuakes):
- choice = input("Sort by (m)agnitude, (t)ime, (l)ongitude, or l(a)titude? ").lower()
- print("")
- if choice == "m":
- allQuakes.sort(key=attrgetter("mag"), reverse=True)
- if choice == "t":
- allQuakes.sort(key=attrgetter("time"), reverse=True)
- if choice == "l":
- allQuakes.sort(key=attrgetter("longitude"), reverse=True)
- if choice == "a":
- allQuakes.sort(key=attrgetter("latitude"))
- printEarthquake(allQuakes)
- def choice_F(allQuakes):
- choice = input("Filter by (m)agnitude or (p)lace? ").lower()
- # print("")
- if choice == "m":
- low = float(input("Lower bound: "))
- high = float(input("Upper bound: "))
- print("")
- printEarthquake(filter_by_mag(allQuakes, low, high))
- if choice == "p":
- word = input("Search for what string? ").lower()
- print("")
- printEarthquake(filter_by_place(allQuakes, word))
- def choice_N(allQuakes):
- EQD = get_json('http://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/1.0_hour.geojson')
- # print(EQD["features"][0])
- featList = EQD["features"]
- oldListLength = len(allQuakes)
- # print(featList[0])
- for x in range(len(featList)):
- newQuake = quake_from_feature(featList[x])
- if newQuake not in allQuakes:
- allQuakes.append(newQuake)
- newListLength = len(allQuakes)
- if newListLength > oldListLength:
- print("nNew quakes found!!!n")
- printEarthquake(allQuakes)
- def write_quakes_to_file(filename, allQuakes):
- out_file = open(filename, "w")
- for eq in earthquakes:
- out_file.write(f"{eq.mag} {eq.longitude} {eq.latitude} {eq.time} {eq.place}n")
- out_file.close()
- # newQuake = quake_from_feature(f)
- # if newQuake == True and (newQuake in allQuakes == False):
- # allQuakes.append(newQuake)
- # print("New quakes found!!!")
- # printEarthquake(allQuakes)
- # valX = read_quakes_from_file("quakes.txt")
- # # printEarthquake(valX)
- # printEarthquake(filter_by_place(valX, "ca"))
- # print(read_quakes_from_file("quakes.txt"))
- --- MAIN BELOW ---
- from quakeFuncs import *
- def main():
- # print("Earthquakes:")
- # print("------------")
- feature = get_json("http://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/1.0_hour.geojson")
- print("")
- allQuakes = read_quakes_from_file("quakes.txt")
- printEarthquake(allQuakes)
- choice = getChoice()
- while choice != "q":
- if choice == "s":
- choice_S(allQuakes)
- if choice == "f":
- choice_F(allQuakes)
- if choice == "n":
- choice_N(allQuakes)
- choice = getChoice()
- if choice == "q":
- write_quakes_to_file("quakes.txt", allQuakes)
- quit()
- if __name__ == "__main__":
- main()