A script to check air pollution in Singapore.

checkpollutants.py

"""
A script that gets the most current PM2.5 and
PSI (Pollutant Standards Index) levels from
data.gov.sg.

Licensed under the MIT License.
Copyright (c) 2019 sn3ksoftware
"""

import requests
from datetime import datetime

# Check if running in StaSh.
try:
	_stash = globals()['_stash']
except KeyError:
	ON_STASH = True
else:
	ON_STASH = True

display = """
This data is under the terms of the
Singapore Open Data License:
(https://data.gov.sg/open-data-licence)

Levels of pollutants:
        PSI    PM2.5
North   {}     {}
South   {}     {}
East    {}     {}
West    {}     {}
Average {}     {}

"""

date_attributes = """
Data recorded on {}, at {}.
"""


def ansi(code_r, bold=""):
	"""ANSI Colours for printing (Because why not?)
	Code can be 30-37. In order of colours,
	these are black, red, green, yellow,
	blue, magnenta, cyan, and white.
	After every colour print, print ansi(0) to clear colour attributes.
	"""
	ansi_base = "\033"
	code = str(code_r)
	if bold:
		ansi_code = ansi_base + "[" + code + ";1m"
		return ansi_code
	else:
		ansi_code = ansi_base + "[" + code + "m"
		return ansi_code


def parse_timestamp(timestamp):
	"""Parses the timestamp in the format
	YYYY-MM-DDThh:mm:ss+/-timezone
	"""
	date, time_r = timestamp.split("T")
	timezone = time_r[-6:]
	time = time_r[:8]
	return [date, time, timezone]


def get_latest_data():
	"""Get the latest PSI and PM2.5 levels
	"""
	# Get current date
	current_date = datetime.today().strftime('%Y-%m-%d')
	# Download from the data.gov.sg API
	gov_data = requests.get(
		"https://api.data.gov.sg/v1/environment/psi?date={}".format(current_date)
		).json()
	# Take the latest reading and timestamp
	latest_data = gov_data["items"][-1]
	timestamp = latest_data["timestamp"]
	readings = latest_data["readings"]
	return [timestamp, readings]


def main():
	# Get current date
	today = current_date = datetime.today().strftime('%d-%m-%Y')
	# Get data using get_latest_data
	readtime, data = get_latest_data()
	pm25 = data["pm25_twenty_four_hourly"]
	psi = data["psi_twenty_four_hourly"]
	# Get individual numbers for each region
	data_combined = {
		"ps_n": psi["north"],
		"pm_n": pm25["north"],
		"ps_s": psi["south"],
		"pm_s": pm25["south"],
		"ps_e": psi["east"],
		"pm_e": pm25["east"],
		"ps_w": psi["west"],
		"pm_w": pm25["west"],
		"ps_a": psi["national"],
		"pm_a": pm25["national"]
	}
	# Add ANSI colour codes if in StaSh.
	for key, val in data_combined.items():
		if not ON_STASH:
			break
		else:
			pass

		if key.startswith("ps"):
			# Check the level and add appropiate colors
			if val <= 50:
				new_val = ansi(32, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 50 and val <= 100:
				new_val = ansi(36, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 101 and val <= 200:
				new_val = ansi(34, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 201 and val <= 300:
				new_val = ansi(35, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 300:
				new_val = ansi(31, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
		elif key.startswith("pm"):
			# Check the level and add appropiate colors
			if val <= 12:
				new_val = ansi(32, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 13 and val <= 55:
				new_val = ansi(36, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 56 and val <= 150:
				new_val = ansi(34, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 151 and val <= 250:
				new_val = ansi(35, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
			elif val > 251:
				new_val = ansi(31, bold=True) + str(val) + ansi(0)
				data_combined.update({key: new_val})
	user_display = display.format(
		*(data_combined.values())
		)
	print(user_display)

if __name__ == "__main__":
	main()