diff --git a/scl.py b/scl.py index b67bc60..4ff759d 100644 --- a/scl.py +++ b/scl.py @@ -75,7 +75,7 @@ class AttribStaticMap(StaticMap, object): image.paste(txt, (0, 0), txt) -def classify_event_size(num_people: int) -> dict[str, str, bool]: +def classify_event_size(num_people: int) -> dict[str, str | bool]: if num_people < 250: return { "size": "Small", @@ -96,6 +96,161 @@ def classify_event_size(num_people: int) -> dict[str, str, bool]: } +def get_hashtag_string(event) -> str: + hashtag_string = "#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format( + event["identifier"] + ) + return hashtag_string + + +def do_initial_post( + event, event_class, start_time: datetime, estimated_restoration_time: datetime +) -> dict[str, str | None]: + post_id = None + map_media_post_id = None + # Fallback location from the SCL API in case one couldn't be reverse geocoded + 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, + # Appending 7F to the fill_color makes it 50% transparent + fill_color="{}7F".format(event_class["outage_color"]), + outline_color=event_class["outage_color"], + simplify=True, + ) + map.add_polygon(polygon) + map_image = map.render() + + try: + + def num2deg(xtile, ytile, zoom): + n = 1 << zoom + lon_deg = xtile / n * 360.0 - 180.0 + lat_rad = math.atan(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"] != "Seattle": + city_not_seattle_text = " of {}".format( + geocode["features"][0]["properties"]["geocoding"]["city"] + ) + else: + city_not_seattle_text = "" + + street = geocode["features"][0]["properties"]["geocoding"]["name"] + + if ( + "locality" in geocode["features"][0]["properties"]["geocoding"] + and event_class["size"] != "Large" + ): + 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( + street, + locality, + city_not_seattle_text, + ) + area_text = "the {} area{}".format(locality, city_not_seattle_text) + elif "district" in geocode["features"][0]["properties"]["geocoding"]: + alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format( + street, + geocode["features"][0]["properties"]["geocoding"]["district"], + city_not_seattle_text, + ) + area_text = "the {} area{}".format( + geocode["features"][0]["properties"]["geocoding"]["district"], + city_not_seattle_text, + ) + else: + alt_text = "A map showing the location of the outage, centered around {} in {}.".format( + street, + geocode["features"][0]["properties"]["geocoding"]["city"], + ) + area_text = geocode["features"][0]["properties"]["geocoding"]["city"] + except Exception: + alt_text = "A map showing the location of the outage." + + with io.BytesIO() as map_image_file: + map_image.save(map_image_file, format="PNG", optimize=True) + map_media_post = mastodon_client.media_post( + map_image_file.getvalue(), + mime_type="image/png", + description=alt_text, + ) + map_media_post_id = map_media_post["id"] + + except Exception as e: + print(e) + print( + "Ran into an issue with generating/uploading the map. Will post without it." + ) + map_media_post = None + hashtag_string = get_hashtag_string(event) + + post_text = """Seattle City Light is reporting a {} outage in {}. + +Start Date: {} +Est. Restoration: {} +Cause: {} + +{}""".format( + event_class["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 + ) + ) + + post_result = mastodon_client.status_post( + status=post_text, + media_ids=map_media_post, + visibility="public", + language="en", + ) + post_id = post_result["id"] + + return {"post_id": post_id, "map_media_post_id": map_media_post_id} + + class Base(DeclarativeBase): pass @@ -142,11 +297,8 @@ with Session(engine) as session: else: status = None - hashtag_string = "#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format( - event["identifier"] - ) - try: + hashtag_string = get_hashtag_string(event) existing_record = lookup_result.one() updated_properties = [] updated_entries = [] @@ -208,9 +360,16 @@ with Session(engine) as session: existing_record.most_recent_post_id = post_result["id"] elif max_event_class["is_postable"]: print( - "This event would have been able to be posted, but someone didn't write the logic to do that for events that scaled up." + "Posting an event that grew above the threshold required to post" ) - + initial_post_result = do_initial_post( + event, event_class, start_time, estimated_restoration_time + ) + existing_record.initial_post_id = initial_post_result["post_id"] + existing_record.most_recent_post_id = initial_post_result["post_id"] + existing_record.map_media_post_id = initial_post_result[ + "map_media_post_id" + ] session.commit() except NoResultFound: @@ -224,174 +383,11 @@ with Session(engine) as session: ) ) else: - # Fallback location from the SCL API in case one couldn't be reverse geocoded - 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, - # Appending 7F to the fill_color makes it 50% transparent - fill_color="{}7F".format(event_class["outage_color"]), - outline_color=event_class["outage_color"], - simplify=True, - ) - map.add_polygon(polygon) - map_image = map.render() - - try: - - def num2deg(xtile, ytile, zoom): - n = 1 << zoom - lon_deg = xtile / n * 360.0 - 180.0 - lat_rad = math.atan( - 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"] - != "Seattle" - ): - city_not_seattle_text = " of {}".format( - geocode["features"][0]["properties"]["geocoding"][ - "city" - ] - ) - else: - city_not_seattle_text = "" - - street = geocode["features"][0]["properties"]["geocoding"][ - "name" - ] - - if ( - "locality" - in geocode["features"][0]["properties"]["geocoding"] - and event_class["size"] != "Large" - ): - 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( - street, - locality, - city_not_seattle_text, - ) - area_text = "the {} area{}".format( - locality, city_not_seattle_text - ) - elif ( - "district" - in geocode["features"][0]["properties"]["geocoding"] - ): - alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format( - street, - geocode["features"][0]["properties"]["geocoding"][ - "district" - ], - city_not_seattle_text, - ) - area_text = "the {} area{}".format( - geocode["features"][0]["properties"]["geocoding"][ - "district" - ], - city_not_seattle_text, - ) - else: - alt_text = "A map showing the location of the outage, centered around {} in {}.".format( - street, - geocode["features"][0]["properties"]["geocoding"][ - "city" - ], - ) - area_text = geocode["features"][0]["properties"][ - "geocoding" - ]["city"] - except Exception: - alt_text = "A map showing the location of the outage." - - with io.BytesIO() as map_image_file: - map_image.save(map_image_file, format="PNG", optimize=True) - map_media_post = mastodon_client.media_post( - map_image_file.getvalue(), - mime_type="image/png", - description=alt_text, - ) - map_media_post_id = map_media_post["id"] - - 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( - event_class["size"].lower(), - area_text, - start_time.strftime(post_datetime_format), - estimated_restoration_time.strftime(post_datetime_format), - event["cause"], - hashtag_string, + initial_post_result = do_initial_post( + event, event_class, start_time, estimated_restoration_time ) - - print( - "Posting the following to Mastodon, with a post length of {}:\n{}".format( - len(post_text), post_text - ) - ) - - post_result = mastodon_client.status_post( - status=post_text, - media_ids=map_media_post, - visibility="public", - language="en", - ) - post_id = post_result["id"] + post_id = initial_post_result["post_id"] + map_media_post_id = initial_post_result["map_media_post_id"] new_outage_record = SclOutage( scl_outage_id=event["id"],