buckbanzai/seattlecitylight-mastodon-bot
1
0
Fork 0
Code Issues 7 Pull requests Projects Releases Packages Wiki Activity

Log to DB, even if it's a small outage

Browse source
This commit is contained in:
Liam Steckler 2024-01-29 22:42:12 -05:00
parent b5bb71e099
commit ea5d52494e
1 changed files with 153 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

289
scl.py
View file

@ -89,9 +89,10 @@ class SclOutage(Base):
outage_size: Mapped[str] = mapped_column() outage_size: Mapped[str] = mapped_column()
status: Mapped[Optional[str]] = mapped_column() status: Mapped[Optional[str]] = mapped_column()
no_longer_in_response_time: Mapped[Optional[datetime]] = mapped_column() no_longer_in_response_time: Mapped[Optional[datetime]] = mapped_column()
start_time: Mapped[datetime] = mapped_column()
def __repr__(self) -> str: def __repr__(self) -> str:
return f"SclOutage(scl_outage_id={self.scl_outage_id!r}, most_recent_post_id={self.most_recent_post_id!r}, last_updated_time={self.last_updated_time!r}, no_longer_in_response_time={self.no_longer_in_response_time!r})" return f"SclOutage(scl_outage_id={self.scl_outage_id!r}, most_recent_post_id={self.most_recent_post_id!r}, last_updated_time={self.last_updated_time!r}, no_longer_in_response_time={self.no_longer_in_response_time!r}), start_time={self.start_time!r})"
engine = create_engine("sqlite:///scl.db") engine = create_engine("sqlite:///scl.db")
@ -183,162 +184,177 @@ with Session(engine) as session:
session.commit() session.commit()
except NoResultFound: except NoResultFound:
print("Existing record not found")
if outage_size == "Small": if outage_size == "Small":
# If the outage becomes medium/large, it'll then be posted as a new outage on the next run # If the outage becomes medium/large, it'll then be posted as a new outage on the next run
print("Outage is small, will not post") print("Outage is small, will not post")
continue continue
print("Existing record not found") else:
# Fallback location from the SCL API in case one couldn't be reverse geocoded
# Fallback location from the SCL API in case one couldn't be reverse geocoded area_text = event["city"]
area_text = event["city"]
try:
map = AttribStaticMap(
512,
512,
url_template="https://tiles.stadiamaps.com/tiles/outdoors/{z}/{x}/{y}.png?api_key="
+ stadiamaps_api_key,
)
assert event["polygons"]["type"] == "polygon"
for ring in event["polygons"]["rings"]:
polygon = Polygon(
ring, "{}7F".format(outage_color), outage_color, simplify=True
)
map.add_polygon(polygon)
map_image = map.render()
try: try:
map = AttribStaticMap(
def num2deg(xtile, ytile, zoom): 512,
n = 1 << zoom 512,
lon_deg = xtile / n * 360.0 - 180.0 url_template="https://tiles.stadiamaps.com/tiles/outdoors/{z}/{x}/{y}.png?api_key="
lat_rad = math.atan(math.sinh(math.pi * (1 - 2 * ytile / n))) + stadiamaps_api_key,
lat_deg = math.degrees(lat_rad)
return lat_deg, lon_deg
center_lat_lon = num2deg(map.x_center, map.y_center, map.zoom)
# Check to make sure the calculated lat and lon are sane enough
# NW Corner
assert center_lat_lon[0] < 48 and center_lat_lon[1] > -122.6
# SE Corner
assert center_lat_lon[0] > 47.2 and center_lat_lon[1] < -122
# Zoom level 17 ensures that we won't get any building/POI names, just street names
geocode_url = "{nominatim_url}/reverse?lat={lat}&lon={lon}&format=geocodejson&zoom=17".format(
nominatim_url=nominatim_url,
lat=center_lat_lon[0],
lon=center_lat_lon[1],
)
geocode_headers = {"User-Agent": "seattlecitylight-mastodon-bot"}
geocode_response = requests.get(
geocode_url, headers=geocode_headers
) )
assert event["polygons"]["type"] == "polygon"
for ring in event["polygons"]["rings"]:
polygon = Polygon(
ring,
"{}7F".format(outage_color),
outage_color,
simplify=True,
)
map.add_polygon(polygon)
map_image = map.render()
try: try:
geocode = geocode_response.json()
except requests.JSONDecodeError:
print("JSON could not be loaded from nominatim API")
raise
if ( def num2deg(xtile, ytile, zoom):
geocode["features"][0]["properties"]["geocoding"]["city"] n = 1 << zoom
!= "Seattle" lon_deg = xtile / n * 360.0 - 180.0
): lat_rad = math.atan(
city_not_seattle_text = " of {}".format( math.sinh(math.pi * (1 - 2 * ytile / n))
)
lat_deg = math.degrees(lat_rad)
return lat_deg, lon_deg
center_lat_lon = num2deg(map.x_center, map.y_center, map.zoom)
# Check to make sure the calculated lat and lon are sane enough
# NW Corner
assert center_lat_lon[0] < 48 and center_lat_lon[1] > -122.6
# SE Corner
assert center_lat_lon[0] > 47.2 and center_lat_lon[1] < -122
# Zoom level 17 ensures that we won't get any building/POI names, just street names
geocode_url = "{nominatim_url}/reverse?lat={lat}&lon={lon}&format=geocodejson&zoom=17".format(
nominatim_url=nominatim_url,
lat=center_lat_lon[0],
lon=center_lat_lon[1],
)
geocode_headers = {
"User-Agent": "seattlecitylight-mastodon-bot"
}
geocode_response = requests.get(
geocode_url, headers=geocode_headers
)
try:
geocode = geocode_response.json()
except requests.JSONDecodeError:
print("JSON could not be loaded from nominatim API")
raise
if (
geocode["features"][0]["properties"]["geocoding"]["city"] geocode["features"][0]["properties"]["geocoding"]["city"]
) != "Seattle"
else: ):
city_not_seattle_text = "" city_not_seattle_text = " of {}".format(
geocode["features"][0]["properties"]["geocoding"][
"city"
]
)
else:
city_not_seattle_text = ""
street = geocode["features"][0]["properties"]["geocoding"]["name"] street = geocode["features"][0]["properties"]["geocoding"][
"name"
]
if ( if (
"locality" in geocode["features"][0]["properties"]["geocoding"]
and outage_size != "Large"
):
locality = geocode["features"][0]["properties"]["geocoding"][
"locality" "locality"
] in geocode["features"][0]["properties"]["geocoding"]
if locality == "Uptown": and outage_size != "Large"
locality = "Lower Queen Anne" ):
locality = geocode["features"][0]["properties"][
"geocoding"
]["locality"]
if locality == "Uptown":
locality = "Lower Queen Anne"
alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format( alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
street, street,
locality, locality,
city_not_seattle_text, city_not_seattle_text,
) )
area_text = "the {} area{}".format( area_text = "the {} area{}".format(
locality, city_not_seattle_text locality, city_not_seattle_text
) )
elif ( elif (
"district" in geocode["features"][0]["properties"]["geocoding"] "district"
): in geocode["features"][0]["properties"]["geocoding"]
alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format( ):
street, alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
geocode["features"][0]["properties"]["geocoding"][ street,
"district" geocode["features"][0]["properties"]["geocoding"][
], "district"
city_not_seattle_text, ],
) city_not_seattle_text,
area_text = "the {} area{}".format( )
geocode["features"][0]["properties"]["geocoding"][ area_text = "the {} area{}".format(
"district" geocode["features"][0]["properties"]["geocoding"][
], "district"
city_not_seattle_text, ],
) city_not_seattle_text,
else: )
alt_text = "A map showing the location of the outage, centered around {} in {}.".format( else:
street, alt_text = "A map showing the location of the outage, centered around {} in {}.".format(
geocode["features"][0]["properties"]["geocoding"]["city"], street,
) geocode["features"][0]["properties"]["geocoding"][
area_text = geocode["features"][0]["properties"]["geocoding"][ "city"
"city" ],
] )
except Exception: area_text = geocode["features"][0]["properties"][
alt_text = "A map showing the location of the outage." "geocoding"
]["city"]
except Exception:
alt_text = "A map showing the location of the outage."
with io.BytesIO() as map_image_file: with io.BytesIO() as map_image_file:
map_image.save(map_image_file, format="PNG", optimize=True) map_image.save(map_image_file, format="PNG", optimize=True)
map_media_post = mastodon.media_post( map_media_post = mastodon.media_post(
map_image_file.getvalue(), map_image_file.getvalue(),
mime_type="image/png", mime_type="image/png",
description=alt_text, description=alt_text,
)
except Exception as e:
print(e)
print(
"Ran into an issue with generating/uploading the map. Will post without it."
) )
map_media_post = None
post_text = """Seattle City Light is reporting a {} outage in {}.
Start Date: {}
Est. Restoration: {}
Cause: {}
{}""".format(
outage_size.lower(),
area_text,
start_time.strftime(post_datetime_format),
estimated_restoration_time.strftime(post_datetime_format),
event["cause"],
hashtag_string,
)
except Exception as e:
print(e)
print( print(
"Ran into an issue with generating/uploading the map. Will post without it." "Posting the following to Mastodon, with a post length of {}:\n{}".format(
len(post_text), post_text
)
) )
map_media_post = None
post_text = """Seattle City Light is reporting a {} outage in {}. mastodon_post_result = mastodon.status_post(
status=post_text,
Start Date: {} media_ids=map_media_post,
Est. Restoration: {} visibility="public",
Cause: {} language="en",
{}""".format(
outage_size.lower(),
area_text,
start_time.strftime(post_datetime_format),
estimated_restoration_time.strftime(post_datetime_format),
event["cause"],
hashtag_string,
)
print(
"Posting the following to Mastodon, with a post length of {}:\n{}".format(
len(post_text), post_text
) )
)
mastodon_post_result = mastodon.status_post(
status=post_text,
media_ids=map_media_post,
visibility="public",
language="en",
)
new_outage_record = SclOutage( new_outage_record = SclOutage(
scl_outage_id=event["id"], scl_outage_id=event["id"],
@ -349,6 +365,7 @@ Cause: {}
cause=event["cause"], cause=event["cause"],
status=status, status=status,
outage_size=outage_size, outage_size=outage_size,
start_time=start_time,
) )
session.add(new_outage_record) session.add(new_outage_record)
session.commit() session.commit()