diff --git a/scl.py b/scl.py index 26cac52..44f26e3 100644 --- a/scl.py +++ b/scl.py @@ -89,9 +89,10 @@ class SclOutage(Base): outage_size: Mapped[str] = mapped_column() status: Mapped[Optional[str]] = mapped_column() no_longer_in_response_time: Mapped[Optional[datetime]] = mapped_column() + start_time: Mapped[datetime] = mapped_column() 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") @@ -183,162 +184,177 @@ with Session(engine) as session: session.commit() except NoResultFound: + print("Existing record not found") if outage_size == "Small": # 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") continue - print("Existing record not found") - - # 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, "{}7F".format(outage_color), outage_color, simplify=True - ) - map.add_polygon(polygon) - map_image = map.render() + else: + # Fallback location from the SCL API in case one couldn't be reverse geocoded + area_text = event["city"] 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 + 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: - 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( + 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"] - ) - else: - city_not_seattle_text = "" + != "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"] + street = geocode["features"][0]["properties"]["geocoding"][ + "name" + ] - if ( - "locality" in geocode["features"][0]["properties"]["geocoding"] - and outage_size != "Large" - ): - locality = geocode["features"][0]["properties"]["geocoding"][ + if ( "locality" - ] - if locality == "Uptown": - locality = "Lower Queen Anne" + in geocode["features"][0]["properties"]["geocoding"] + and outage_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." + 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.media_post( - map_image_file.getvalue(), - mime_type="image/png", - description=alt_text, + with io.BytesIO() as map_image_file: + map_image.save(map_image_file, format="PNG", optimize=True) + map_media_post = mastodon.media_post( + map_image_file.getvalue(), + mime_type="image/png", + 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( - "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 {}. - -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, - ) - - 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", ) - ) - - mastodon_post_result = mastodon.status_post( - status=post_text, - media_ids=map_media_post, - visibility="public", - language="en", - ) new_outage_record = SclOutage( scl_outage_id=event["id"], @@ -349,6 +365,7 @@ Cause: {} cause=event["cause"], status=status, outage_size=outage_size, + start_time=start_time, ) session.add(new_outage_record) session.commit()