diff --git a/scl.py b/scl.py index 4ff759d..b67bc60 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,161 +96,6 @@ 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 @@ -297,8 +142,11 @@ 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 = [] @@ -360,16 +208,9 @@ with Session(engine) as session: existing_record.most_recent_post_id = post_result["id"] elif max_event_class["is_postable"]: print( - "Posting an event that grew above the threshold required to post" + "This event would have been able to be posted, but someone didn't write the logic to do that for events that scaled up." ) - 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: @@ -383,11 +224,174 @@ with Session(engine) as session: ) ) else: - initial_post_result = do_initial_post( - event, event_class, start_time, estimated_restoration_time + # 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, ) - post_id = initial_post_result["post_id"] - map_media_post_id = initial_post_result["map_media_post_id"] + + 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"] new_outage_record = SclOutage( scl_outage_id=event["id"],