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

Don't store size classes in the DB

Browse source
This commit is contained in:
Liam Steckler 2024-01-30 07:58:11 -08:00
parent e402a86670
commit 1e11345a4b
1 changed files with 41 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
scl.py
View file

@ -74,6 +74,27 @@ class AttribStaticMap(StaticMap, object):
image.paste(txt, (0, 0), txt)
def classify_event_size(num_people: int) -> dict[str, str, bool]:
if num_people < 250:
return {
"size": "Small",
"outage_color": "#F97316",
"is_postable": False,
}
elif event["numPeople"] < 1000:
return {
"size": "Medium",
"outage_color": "#EF4444",
"is_postable": True,
}
else:
return {
"size": "Large",
"outage_color": "#991B1B",
"is_postable": True,
}
class Base(DeclarativeBase):
pass
@ -88,14 +109,14 @@ class SclOutage(Base):
last_updated_time: Mapped[datetime] = mapped_column()
estimated_restoration_time: Mapped[datetime] = mapped_column()
cause: Mapped[str] = mapped_column()
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()
num_people: Mapped[int] = mapped_column()
max_num_people: Mapped[int] = 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}, initial_post_id={self.initial_post_id!r}, map_media_post_id={self.map_media_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}), num_people={self.num_people!r})"
return f"SclOutage(scl_outage_id={self.scl_outage_id!r}, most_recent_post_id={self.most_recent_post_id!r}, initial_post_id={self.initial_post_id!r}, map_media_post_id={self.map_media_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}), num_people={self.num_people!r}), max_num_people={self.max_num_people!r})"
engine = create_engine("sqlite:///scl.db")
@ -112,15 +133,8 @@ with Session(engine) as session:
SclOutage.scl_outage_id == event["id"]
)
lookup_result = session.scalars(lookup_statement)
if event["numPeople"] < 250:
outage_size = "Small"
outage_color = "#F97316"
elif event["numPeople"] < 1000:
outage_size = "Medium"
outage_color = "#EF4444"
else:
outage_size = "Large"
outage_color = "#991B1B"
event_class = classify_event_size(event["numPeople"])
if "status" in event:
status = event["status"]
@ -147,16 +161,20 @@ with Session(engine) as session:
existing_record.cause = event["cause"]
updated_properties.append("cause")
updated_entries.append("Cause: {}".format(event["cause"]))
if outage_size != existing_record.outage_size:
existing_record.outage_size = outage_size
previous_event_class = classify_event_size(existing_record.num_people)
if event_class["size"] != previous_event_class["size"]:
updated_properties.append("outage size")
updated_entries.append("Outage Size: {}".format(outage_size))
updated_entries.append("Outage Size: {}".format(event_class["size"]))
if status != existing_record.status:
existing_record.status = status
updated_properties.append("status")
updated_entries.append("Status: {}".format(status))
if existing_record.num_people != event["numPeople"]:
existing_record.num_people = event["numPeople"]
if existing_record.max_num_people < event["numPeople"]:
# Used to determine the maximum number of people affected by this outage, to determine if it's worth posting about
existing_record.max_num_people = event["numPeople"]
if updated_properties:
updated_properties.sort()
@ -190,9 +208,8 @@ with Session(engine) as session:
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")
if not event_class["is_postable"]:
print("Outage is not considered postable, will not post")
continue
else:
# Fallback location from the SCL API in case one couldn't be reverse geocoded
@ -209,8 +226,9 @@ with Session(engine) as session:
for ring in event["polygons"]["rings"]:
polygon = Polygon(
ring,
"{}7F".format(outage_color),
outage_color,
# 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)
@ -272,7 +290,7 @@ with Session(engine) as session:
if (
"locality"
in geocode["features"][0]["properties"]["geocoding"]
and outage_size != "Large"
and event_class["size"] != "Large"
):
locality = geocode["features"][0]["properties"][
"geocoding"
@ -340,7 +358,7 @@ with Session(engine) as session:
Cause: {}
{}""".format(
outage_size.lower(),
event_class["size"].lower(),
area_text,
start_time.strftime(post_datetime_format),
estimated_restoration_time.strftime(post_datetime_format),
@ -366,14 +384,14 @@ with Session(engine) as session:
outage_user_id=event["identifier"],
most_recent_post_id=mastodon_post_result["id"],
initial_post_id=mastodon_post_result["id"],
map_media_post_id=map_media_post["id"],
last_updated_time=last_updated_time,
estimated_restoration_time=estimated_restoration_time,
cause=event["cause"],
status=status,
outage_size=outage_size,
start_time=start_time,
num_people=event["numPeople"],
map_media_post_id=map_media_post["id"],
max_num_people=event["numPeople"],
)
session.add(new_outage_record)
session.commit()