#!/usr/bin/env python
# flake8: noqa
from __future__ import annotations
import re
import unicodedata
from datetime import datetime
from typing import Dict
from fireo import fields
from fireo.models import Model
from ..utils import file_utils
from . import validators
from .constants import (
EventMinutesItemDecision,
MatterStatusDecision,
Order,
RoleTitle,
VoteDecision,
)
from .types import IndexedField, IndexedFieldSet
###############################################################################
[docs]
class File(Model):
"""A file from the CDP file store."""
id = fields.IDField()
uri = fields.TextField(required=True, validator=validators.resource_exists)
name = fields.TextField(required=True)
description = fields.TextField()
media_type = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
required=True, validator=validators.resource_exists, validator_kwargs=kwargs
)
field.contribute_to_model(File, "uri")
[docs]
@classmethod
def Example(cls) -> Model:
uri = "gs://cdp-example/central-staff-memo.pdf"
file = cls()
file.uri = uri
file.name = "Central Staff Memo"
file.media_type = file_utils.get_media_type(uri)
return file
_PRIMARY_KEYS = ("uri",)
_INDEXES = ()
[docs]
class Person(Model):
"""
Primarily the council members, this could technically include the mayor or city
manager, or any other "normal" presenters and attendees of meetings.
"""
id = fields.IDField()
name = fields.TextField(required=True)
router_string = fields.TextField(
required=True, validator=validators.router_string_is_valid
)
email = fields.TextField(validator=validators.email_is_valid)
phone = fields.TextField()
website = fields.TextField(validator=validators.resource_exists)
picture_ref = fields.ReferenceField(File, auto_load=False)
is_active = fields.BooleanField(required=True)
external_source_id = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
validator=validators.resource_exists,
validator_kwargs=kwargs,
)
field.contribute_to_model(Person, "website")
[docs]
@classmethod
def Example(cls) -> Model:
person = cls()
person.name = "M. Lorena González"
person.router_string = "lorena-gonzalez"
person.is_active = True
return person
_PRIMARY_KEYS = ("name",)
_INDEXES = ()
[docs]
@staticmethod
def strip_accents(name: str) -> str:
return "".join(
char
for char in unicodedata.normalize("NFKD", name)
if unicodedata.category(char) != "Mn"
)
[docs]
@staticmethod
def generate_router_string(name: str) -> str:
non_accented = Person.strip_accents(name)
char_cleaner = re.compile(r"[^a-zA-Z0-9\s\-]")
char_cleaned = re.sub(char_cleaner, "", non_accented)
whitespace_cleaner = re.compile(r"[\s]+")
whitespace_cleaned = re.sub(whitespace_cleaner, " ", char_cleaned)
spaces_replaced = whitespace_cleaned.replace(" ", "-")
if spaces_replaced[-1] == "-":
fully_cleaned = spaces_replaced[:-1]
else:
fully_cleaned = spaces_replaced
return fully_cleaned.lower()
[docs]
class Body(Model):
"""
A meeting body. This can be full council, a subcommittee, or "off-council" matters
such as election debates.
"""
id = fields.IDField()
name = fields.TextField(required=True)
description = fields.TextField()
start_datetime = fields.DateTime(required=True)
end_datetime = fields.DateTime()
is_active = fields.BooleanField(required=True)
external_source_id = fields.TextField()
[docs]
class Meta:
ignore_none_field = False
[docs]
@classmethod
def Example(cls) -> Model:
body = cls()
body.name = "Full Council"
body.is_active = True
body.start_datetime = datetime.utcnow()
return body
_PRIMARY_KEYS = ("name",)
_INDEXES = ()
[docs]
class Seat(Model):
"""An electable office on the City Council. I.E. "Position 9"."""
id = fields.IDField()
name = fields.TextField(required=True)
electoral_area = fields.TextField()
electoral_type = fields.TextField()
image_ref = fields.ReferenceField(File, auto_load=False)
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
seat = cls()
seat.name = "Position 9"
seat.electoral_area = "Citywide"
seat.electoral_type = "at-large"
return seat
_PRIMARY_KEYS = ("name",)
_INDEXES = ()
[docs]
class Role(Model):
"""
A role is a person's job for a period of time in the city council. A person can
(and generally does) have many roles. For example: a person has two terms as city
council member for district four then a term as city council member for a citywide
seat. Roles can also be tied to committee chairs. For example: a council member
spends a term on the transportation committee and then spends a term on the finance
committee.
"""
id = fields.IDField()
title = fields.TextField(
required=True,
validator=validators.create_constant_value_validator(RoleTitle),
)
person_ref = fields.ReferenceField(Person, required=True, auto_load=False)
body_ref = fields.ReferenceField(Body, auto_load=False)
seat_ref = fields.ReferenceField(Seat, required=True, auto_load=False)
start_datetime = fields.DateTime(required=True)
end_datetime = fields.DateTime()
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
role = cls()
role.title = RoleTitle.COUNCILPRESIDENT
role.person_ref = Person.Example()
role.body_ref = Body.Example()
role.seat_ref = Seat.Example()
role.start_datetime = datetime.utcnow()
return role
_PRIMARY_KEYS = ("title", "person_ref", "body_ref", "seat_ref")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="start_datetime", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="start_datetime", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="end_datetime", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="end_datetime", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="seat_ref", order=Order.ASCENDING),
IndexedField(fieldPath="end_datetime", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="title", order=Order.ASCENDING),
IndexedField(fieldPath="start_datetime", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="title", order=Order.ASCENDING),
IndexedField(fieldPath="start_datetime", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="title", order=Order.ASCENDING),
IndexedField(fieldPath="end_datetime", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="title", order=Order.ASCENDING),
IndexedField(fieldPath="end_datetime", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="end_datetime", order=Order.ASCENDING),
IndexedField(fieldPath="seat_ref", order=Order.ASCENDING),
IndexedField(fieldPath="start_datetime", order=Order.DESCENDING),
)
),
)
[docs]
class Matter(Model):
"""
A matter is a specific legislative document.
A bill, resolution, initiative, etc.
"""
id = fields.IDField()
name = fields.TextField(required=True)
matter_type = fields.TextField(required=True)
title = fields.TextField(required=True)
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
matter = cls()
matter.name = "CB 119858"
matter.matter_type = "Council Bill"
matter.title = (
"AN ORDINANCE relating to the financing of the West Seattle Bridge "
"Immediate Response project; creating a fund for depositing proceeds of "
"taxable limited tax general obligation bonds in 2021; authorizing the "
"loan of funds in the amount of $50,000,000 from the Construction and "
"Inspections Fund and $20,000,000 from the REET II Capital Projects Fund "
"to the 2021 LTGO Taxable Bond Fund for early phases of work on the bridge "
"repair and replacement project; amending Ordinance 126000, which adopted "
"the 2020 Budget, including the 2020-2025 Capital Improvement Program "
"(CIP); changing appropriations to the Seattle Department of "
"Transportation; and revising project allocations and spending plans for "
"certain projects in the 2020-2025 CIP."
)
return matter
_PRIMARY_KEYS = ("name",)
_INDEXES = ()
[docs]
class MatterFile(Model):
"""
A document related to a matter.
This file is usually not stored in CDP infrastructure but a remote source.
"""
id = fields.IDField()
matter_ref = fields.ReferenceField(Matter, required=True, auto_load=False)
name = fields.TextField(required=True)
uri = fields.TextField(required=True, validator=validators.resource_exists)
external_source_id = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
required=True, validator=validators.resource_exists, validator_kwargs=kwargs
)
field.contribute_to_model(MatterFile, "uri")
[docs]
@classmethod
def Example(cls) -> Model:
matter_file = cls()
matter_file.matter_ref = Matter.Example()
matter_file.name = "Amendment 3 (Sawant) - Sunset"
matter_file.uri = (
"http://legistar2.granicus.com/seattle/attachments/"
"789a0c9f-dd9c-401b-aaf5-6c67c2a897b0.pdf"
)
return matter_file
_PRIMARY_KEYS = ("matter_ref", "name", "uri")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="matter_ref", order=Order.ASCENDING),
IndexedField(fieldPath="name", order=Order.ASCENDING),
)
),
)
[docs]
class MinutesItem(Model):
"""
An item referenced during a meeting.
This can be a matter but it can be a presentation or budget file, etc.
"""
id = fields.IDField()
name = fields.TextField(required=True)
description = fields.TextField()
matter_ref = fields.ReferenceField(Matter, auto_load=False) # Note optional.
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
minutes_item = cls()
minutes_item.name = "Inf 1656"
minutes_item.description = (
"Roadmap to defunding the Police and investing in community."
)
return minutes_item
_PRIMARY_KEYS = ("name",)
_INDEXES = ()
[docs]
class Event(Model):
"""
An event can be a normally scheduled meeting, a special event such as a press
conference or election debate, and, can be upcoming or historical.
"""
id = fields.IDField()
body_ref = fields.ReferenceField(Body, required=True, auto_load=False)
event_datetime = fields.DateTime(required=True)
static_thumbnail_ref = fields.ReferenceField(File, auto_load=False)
hover_thumbnail_ref = fields.ReferenceField(File, auto_load=False)
agenda_uri = fields.TextField(validator=validators.resource_exists)
minutes_uri = fields.TextField(validator=validators.resource_exists)
external_source_id = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
validator=validators.resource_exists, validator_kwargs=kwargs
)
field.contribute_to_model(Event, "agenda_uri")
field.contribute_to_model(Event, "minutes_uri")
[docs]
@classmethod
def Example(cls) -> Model:
event = cls()
event.body_ref = Body.Example()
event.event_datetime = datetime.utcnow()
event.agenda_uri = (
"http://legistar2.granicus.com/seattle/meetings/2019/11/"
"4169_A_Select_Budget_Committee_19-11-01_Committee_Agenda.pdf"
)
event.minutes_uri = (
"http://legistar2.granicus.com/seattle/meetings/2019/7/"
"4041_M_Council_Briefing_19-07-22_Committee_Minutes.pdf"
)
return event
_PRIMARY_KEYS = ("body_ref", "event_datetime")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="body_ref", order=Order.ASCENDING),
IndexedField(fieldPath="event_datetime", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="body_ref", order=Order.ASCENDING),
IndexedField(fieldPath="event_datetime", order=Order.DESCENDING),
)
),
)
[docs]
class Session(Model):
"""
A session is a working period for an event.
For example, An event could have a morning and afternoon session.
"""
id = fields.IDField()
event_ref = fields.ReferenceField(Event, required=True, auto_load=False)
session_datetime = fields.DateTime(required=True)
session_index = fields.NumberField(required=True)
session_content_hash = fields.TextField(required=True)
video_uri = fields.TextField(required=True, validator=validators.resource_exists)
video_start_time = fields.TextField(validators.time_duration_is_valid)
video_end_time = fields.TextField(validators.time_duration_is_valid)
caption_uri = fields.TextField(validator=validators.resource_exists)
external_source_id = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
required=True, validator=validators.resource_exists, validator_kwargs=kwargs
)
field.contribute_to_model(Session, "video_uri")
[docs]
@classmethod
def Example(cls) -> Model:
session = cls()
session.event_ref = Event.Example()
session.session_index = 0
session.video_uri = (
"https://video.seattle.gov/media/council/brief_072219_2011957V.mp4"
)
session.video_start_time = "01:00:00"
session.video_end_time = "99:59:59"
session.session_content_hash = (
"05bd857af7f70bf51b6aac1144046973bf3325c9101a554bc27dc9607dbbd8f5"
)
return session
_PRIMARY_KEYS = ("event_ref", "video_uri")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(fieldPath="session_index", order=Order.ASCENDING),
)
),
)
[docs]
class Transcript(Model):
"""A transcript is a document per-session."""
id = fields.IDField()
session_ref = fields.ReferenceField(Session, required=True, auto_load=False)
file_ref = fields.ReferenceField(File, required=True, auto_load=False)
generator = fields.TextField(required=True)
confidence = fields.NumberField(required=True)
created = fields.DateTime(required=True)
[docs]
@classmethod
def Example(cls) -> Model:
transcript = cls()
transcript.session_ref = Session.Example()
transcript.file_ref = File.Example()
transcript.generator = "FakeGen -- v0.1.0"
transcript.confidence = 0.943
transcript.created = datetime.utcnow()
return transcript
_PRIMARY_KEYS = ("session_ref", "file_ref")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="session_ref", order=Order.ASCENDING),
IndexedField(fieldPath="created", order=Order.DESCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="session_ref", order=Order.ASCENDING),
IndexedField(fieldPath="confidence", order=Order.DESCENDING),
)
),
)
[docs]
class EventMinutesItem(Model):
"""A reference tying a specific minutes item to a specific event."""
id = fields.IDField()
event_ref = fields.ReferenceField(Event, required=True, auto_load=False)
minutes_item_ref = fields.ReferenceField(
MinutesItem, required=True, auto_load=False
)
index = fields.NumberField(required=True)
decision = fields.TextField(
validator=validators.create_constant_value_validator(EventMinutesItemDecision)
)
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
emi = cls()
emi.event_ref = Event.Example()
emi.minutes_item_ref = MinutesItem.Example()
emi.index = 0
emi.decision = EventMinutesItemDecision.PASSED
return emi
_PRIMARY_KEYS = ("event_ref", "minutes_item_ref")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(fieldPath="index", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(fieldPath="index", order=Order.DESCENDING),
)
),
)
[docs]
class MatterStatus(Model):
"""
A matter status is the status of a matter at any given time. Useful for tracking
the timelines of matters. I.E. Return me a timeline of matter x.
The same matter will have multiple matter statuses.
1. MatterStatus of submitted
2. MatterStatus of passed
3. MatterStatus of signed
4. etc.
"""
id = fields.IDField()
matter_ref = fields.ReferenceField(Matter, required=True, auto_load=False)
# Optional because status can be updated out of event
# i.e. Signed by Mayor
event_minutes_item_ref = fields.ReferenceField(EventMinutesItem, auto_load=False)
status = fields.TextField(
required=True,
validator=validators.create_constant_value_validator(MatterStatusDecision),
)
update_datetime = fields.DateTime(required=True)
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
matter_status = cls()
matter_status.matter_ref = Matter.Example()
matter_status.status = MatterStatusDecision.ADOPTED
matter_status.update_datetime = datetime.utcnow()
return matter_status
_PRIMARY_KEYS = ("matter_ref", "status", "update_datetime")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="matter_ref", order=Order.ASCENDING),
IndexedField(fieldPath="update_datetime", order=Order.ASCENDING),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="matter_ref", order=Order.ASCENDING),
IndexedField(fieldPath="update_datetime", order=Order.DESCENDING),
),
),
)
[docs]
class EventMinutesItemFile(Model):
"""Supporting files for an event minutes item."""
id = fields.IDField()
event_minutes_item_ref = fields.ReferenceField(
EventMinutesItem, required=True, auto_load=False
)
name = fields.TextField(required=True)
uri = fields.TextField(required=True, validator=validators.resource_exists)
external_source_id = fields.TextField()
[docs]
def set_validator_kwargs(self, kwargs: Dict) -> None:
field = fields.TextField(
required=True, validator=validators.resource_exists, validator_kwargs=kwargs
)
field.contribute_to_model(EventMinutesItemFile, "uri")
[docs]
@classmethod
def Example(cls) -> Model:
emif = cls()
emif.event_minutes_item_ref = EventMinutesItem.Example()
emif.name = "Levy to Move Seattle Quartly Report"
emif.uri = (
"http://legistar2.granicus.com/seattle/attachments/"
"ec6595cf-e2c3-449d-811b-047675d047df.pdf"
)
return emif
_PRIMARY_KEYS = ("event_minutes_item_ref", "uri")
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="event_minutes_item_ref", order=Order.ASCENDING),
IndexedField(fieldPath="name", order=Order.ASCENDING),
)
),
)
[docs]
class Vote(Model):
"""A reference tying a specific person and an event minutes item together."""
id = fields.IDField()
matter_ref = fields.ReferenceField(Matter, required=True, auto_load=False)
event_ref = fields.ReferenceField(Event, required=True, auto_load=False)
event_minutes_item_ref = fields.ReferenceField(
EventMinutesItem, required=True, auto_load=False
)
person_ref = fields.ReferenceField(Person, required=True, auto_load=False)
decision = fields.TextField(
required=True,
validator=validators.create_constant_value_validator(VoteDecision),
)
in_majority = fields.BooleanField()
external_source_id = fields.TextField()
[docs]
@classmethod
def Example(cls) -> Model:
vote = cls()
vote.matter_ref = Matter.Example()
vote.event_ref = Event.Example()
vote.event_minutes_item_ref = EventMinutesItem.Example()
vote.person_ref = Person.Example()
vote.decision = VoteDecision.APPROVE
vote.in_majority = True
return vote
_PRIMARY_KEYS = (
"matter_ref",
"event_ref",
"event_minutes_item_ref",
"person_ref",
"decision",
)
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="matter_ref", order=Order.ASCENDING),
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
)
),
IndexedFieldSet(
(
IndexedField(fieldPath="person_ref", order=Order.ASCENDING),
IndexedField(fieldPath="matter_ref", order=Order.ASCENDING),
)
),
)
[docs]
class IndexedEventGram(Model):
"""
An n-gram that has already been scored to create, when taken altogether,
an event relevance index.
"""
id = fields.IDField()
event_ref = fields.ReferenceField(Event, required=True, auto_load=False)
unstemmed_gram = fields.TextField(required=True)
stemmed_gram = fields.TextField(required=True)
context_span = fields.TextField(required=True)
value = fields.NumberField(required=True)
datetime_weighted_value = fields.NumberField(required=True)
[docs]
@classmethod
def Example(cls) -> Model:
ieg = cls()
ieg.event_ref = Event.Example()
ieg.unstemmed_gram = "housing"
ieg.stemmed_gram = "hous"
ieg.context_span = "We believe that housing should be affordable."
ieg.value = 12.34
ieg.datetime_weighted_value = 1.234
return ieg
_PRIMARY_KEYS = (
"event_ref",
"unstemmed_gram",
"stemmed_gram",
)
_INDEXES = (
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(fieldPath="value", order=Order.DESCENDING),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="event_ref", order=Order.ASCENDING),
IndexedField(
fieldPath="datetime_weighted_value", order=Order.DESCENDING
),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="stemmed_gram", order=Order.ASCENDING),
IndexedField(fieldPath="value", order=Order.DESCENDING),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="stemmed_gram", order=Order.ASCENDING),
IndexedField(
fieldPath="datetime_weighted_value", order=Order.DESCENDING
),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="unstemmed_gram", order=Order.ASCENDING),
IndexedField(fieldPath="value", order=Order.DESCENDING),
),
),
IndexedFieldSet(
(
IndexedField(fieldPath="unstemmed_gram", order=Order.ASCENDING),
IndexedField(
fieldPath="datetime_weighted_value", order=Order.DESCENDING
),
),
),
)