flake8 compliant
This commit is contained in:
+33
-17
@@ -25,6 +25,11 @@ from src.model_textual import (
|
||||
SalienceModelOutput
|
||||
)
|
||||
|
||||
|
||||
class NoDocumentFoundException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class ModelOutputs(BaseModel):
|
||||
visual_syntax: VisualSyntaxModelOutput
|
||||
contact: ContactModelOutput
|
||||
@@ -52,7 +57,7 @@ class ModelOutputs(BaseModel):
|
||||
in cls.model_fields.items()
|
||||
}
|
||||
return cls(**kwargs)
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_annotations(
|
||||
cls,
|
||||
@@ -70,17 +75,28 @@ class ModelOutputs(BaseModel):
|
||||
) -> ModelOutputs:
|
||||
"""Instantiate from annotation."""
|
||||
kwargs = {
|
||||
"visual_syntax": VisualSyntaxModelOutput.from_choice(visual_syntax),
|
||||
"contact": ContactModelOutput.from_choice(contact),
|
||||
"angle": AngleModelOutput.from_choice(angle),
|
||||
"point_of_view": PointOfViewModelOutput.from_choice(point_of_view),
|
||||
"distance": DistanceModelOutput.from_choice(distance),
|
||||
"modality_lighting": ModalityLightingModelOutput.from_choice(modality_lighting),
|
||||
"modality_color": ModalityColorModelOutput.from_choice(modality_color),
|
||||
"modality_depth": ModalityDepthModelOutput.from_choice(modality_depth),
|
||||
"information_value": InformationValueModelOutput.from_choice(information_value),
|
||||
"framing": FramingModelOutput.from_choice(framing),
|
||||
"salience": SalienceModelOutput.from_choice(salience)
|
||||
"visual_syntax": VisualSyntaxModelOutput
|
||||
.from_choice(visual_syntax),
|
||||
"contact": ContactModelOutput
|
||||
.from_choice(contact),
|
||||
"angle": AngleModelOutput
|
||||
.from_choice(angle),
|
||||
"point_of_view": PointOfViewModelOutput
|
||||
.from_choice(point_of_view),
|
||||
"distance": DistanceModelOutput
|
||||
.from_choice(distance),
|
||||
"modality_lighting": ModalityLightingModelOutput
|
||||
.from_choice(modality_lighting),
|
||||
"modality_color": ModalityColorModelOutput
|
||||
.from_choice(modality_color),
|
||||
"modality_depth": ModalityDepthModelOutput
|
||||
.from_choice(modality_depth),
|
||||
"information_value": InformationValueModelOutput
|
||||
.from_choice(information_value),
|
||||
"framing": FramingModelOutput
|
||||
.from_choice(framing),
|
||||
"salience": SalienceModelOutput
|
||||
.from_choice(salience)
|
||||
}
|
||||
return cls(**kwargs)
|
||||
|
||||
@@ -115,7 +131,10 @@ class VisualCommunication(BaseModel):
|
||||
|
||||
@field_validator("image", mode="before")
|
||||
@classmethod
|
||||
def convert_to_image(cls, image: Image.Image | BytesIO | bytes) -> Image.Image:
|
||||
def convert_to_image(
|
||||
cls,
|
||||
image: Image.Image | BytesIO | bytes
|
||||
) -> Image.Image:
|
||||
if isinstance(image, bytes):
|
||||
image = BytesIO(image)
|
||||
if isinstance(image, BytesIO):
|
||||
@@ -124,7 +143,7 @@ class VisualCommunication(BaseModel):
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.classname()}(name='{self.name}')"
|
||||
|
||||
|
||||
def webencoded_image(self) -> str:
|
||||
"""Convert image to be displayed on webpage."""
|
||||
# convert images to bytes string
|
||||
@@ -138,6 +157,3 @@ class VisualCommunication(BaseModel):
|
||||
if not force and self.prediction is not None:
|
||||
logging.warning("set force=True to overwrite existing values.")
|
||||
self.prediction = ModelOutputs.from_random()
|
||||
|
||||
class NoDocumentFoundException(Exception):
|
||||
pass
|
||||
|
||||
+20
-8
@@ -20,15 +20,17 @@ def total_annotated(
|
||||
) -> int:
|
||||
"""Get total number of annotated documents in database."""
|
||||
query = {
|
||||
"annotation": { "$ne": None }
|
||||
"annotation": {
|
||||
"$ne": None
|
||||
}
|
||||
}
|
||||
return collection.count_documents(filter=query)
|
||||
|
||||
|
||||
def get_visual_communication(
|
||||
collection: Collection,
|
||||
with_annotation: bool = False
|
||||
) -> VisualCommunication:
|
||||
collection: Collection,
|
||||
with_annotation: bool = False
|
||||
) -> VisualCommunication:
|
||||
"""Get a random visual communication from the database."""
|
||||
query = {}
|
||||
if with_annotation:
|
||||
@@ -36,8 +38,14 @@ def get_visual_communication(
|
||||
else:
|
||||
query["annotation"] = None
|
||||
data = collection.aggregate([
|
||||
{ "$match": query }, # find using filters
|
||||
{ "$sample": { "size": 1 } } # get one random
|
||||
{
|
||||
"$match": query # find using filters
|
||||
},
|
||||
{
|
||||
"$sample": {
|
||||
"size": 1 # get one random
|
||||
}
|
||||
}
|
||||
])
|
||||
data = list(data) # read data from cursor object
|
||||
if len(data) == 0:
|
||||
@@ -58,7 +66,9 @@ def upsert_predictions(
|
||||
"name": vis_com_name
|
||||
}
|
||||
update = {
|
||||
"$set": { "prediction": predictions.model_dump() }
|
||||
"$set": {
|
||||
"prediction": predictions.model_dump()
|
||||
}
|
||||
}
|
||||
res = collection.update_one(
|
||||
filter=query,
|
||||
@@ -79,7 +89,9 @@ def upsert_annotations(
|
||||
"name": vis_com_name
|
||||
}
|
||||
update = {
|
||||
"$set": { "annotation": annotations.model_dump() }
|
||||
"$set": {
|
||||
"annotation": annotations.model_dump()
|
||||
}
|
||||
}
|
||||
res = collection.update_one(
|
||||
filter=query,
|
||||
|
||||
+3
-4
@@ -1,9 +1,10 @@
|
||||
import logging
|
||||
from dotenv import load_dotenv
|
||||
from pathlib import Path
|
||||
import logging
|
||||
import os
|
||||
|
||||
from src.web import app
|
||||
|
||||
# prepare optional local setup
|
||||
env_path = Path(__file__).parent.parent / "local.env"
|
||||
load_dotenv(env_path)
|
||||
@@ -38,8 +39,6 @@ datefmt = '%Y-%m-%d %H:%M:%S'
|
||||
logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
|
||||
|
||||
logging.info("initialized app")
|
||||
|
||||
from src.web import app
|
||||
server = app.server
|
||||
|
||||
if __name__ == "__main__":
|
||||
@@ -47,4 +46,4 @@ if __name__ == "__main__":
|
||||
os.environ["MONGO_HOST"] = "localhost"
|
||||
# run app
|
||||
app.run(debug=True)
|
||||
logging.info("started app")
|
||||
logging.info("started app")
|
||||
|
||||
@@ -1,24 +1 @@
|
||||
from .classes import VisualSyntaxModelOutput
|
||||
|
||||
|
||||
|
||||
# CLASS_NAME_LIST = Literal[
|
||||
# "non transactional action",
|
||||
# "non transactional reaction",
|
||||
# "unidirectional transactional action",
|
||||
# "unidirectional transactional reaction",
|
||||
# "bidirectional transactional action",
|
||||
# "bidirectional transactional reaction",
|
||||
# "conversion",
|
||||
# "speech process",
|
||||
# "classification overt taxonomy",
|
||||
# "analytical exhaustive",
|
||||
# "analytical disarranged",
|
||||
# "analytical temporal",
|
||||
# "analytical distributed",
|
||||
# "anaytical topological",
|
||||
# "analytical exploded",
|
||||
# "analytical inclusive",
|
||||
# "symbolic suggestive",
|
||||
# "symbolic attributive"
|
||||
# ]
|
||||
@@ -6,6 +6,7 @@ import random
|
||||
class OptionNotSetException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class ModelOutput(BaseModel):
|
||||
|
||||
@classmethod
|
||||
@@ -17,13 +18,13 @@ class ModelOutput(BaseModel):
|
||||
def list_fields(cls) -> List[str]:
|
||||
"""List options that are stored as attributes."""
|
||||
return list(cls.model_fields.keys())
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_random(cls):
|
||||
"""Instantiate with random numbers."""
|
||||
kwargs = {field: random.random() for field in cls.list_fields()}
|
||||
return cls(**kwargs)
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_choice(cls, option: str):
|
||||
"""Instantiate from choice."""
|
||||
@@ -31,7 +32,8 @@ class ModelOutput(BaseModel):
|
||||
raise ValidationError()
|
||||
assert isinstance(option, str), "option is not a string"
|
||||
allowed_options_list = cls.list_fields()
|
||||
assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
|
||||
assert option in allowed_options_list, \
|
||||
f"{option} is not among allowed fields {allowed_options_list}"
|
||||
kwargs = {field: 0 for field in cls.list_fields()}
|
||||
kwargs[option] = 1
|
||||
return cls(**kwargs)
|
||||
@@ -39,15 +41,19 @@ class ModelOutput(BaseModel):
|
||||
def __repr__(self) -> str:
|
||||
model_dict = self.model_dump()
|
||||
model_repr_str = f"{self.classname()}("
|
||||
model_repr_str += ", ".join([f"{field}={value:.3f}" for field, value in model_dict.items()])
|
||||
model_repr_str += ", ".join([
|
||||
f"{field}={value:.3f}"
|
||||
for field, value
|
||||
in model_dict.items()
|
||||
])
|
||||
model_repr_str += ")"
|
||||
return model_repr_str
|
||||
|
||||
|
||||
def highest_score_field(self) -> str:
|
||||
"""Return name of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
return max(model_dict, key=lambda k: model_dict[k])
|
||||
|
||||
|
||||
def highest_score_value(self) -> float:
|
||||
"""Return value of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
|
||||
@@ -6,4 +6,4 @@ from .classes import (
|
||||
ModalityLightingModelOutput,
|
||||
ModalityColorModelOutput,
|
||||
ModalityDepthModelOutput
|
||||
)
|
||||
)
|
||||
|
||||
@@ -14,13 +14,13 @@ class ModelOutput(BaseModel):
|
||||
def list_fields(cls) -> List[str]:
|
||||
"""List options that are stored as attributes."""
|
||||
return list(cls.model_fields.keys())
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_random(cls):
|
||||
"""Instantiate with random numbers."""
|
||||
kwargs = {field: random.random() for field in cls.list_fields()}
|
||||
return cls(**kwargs)
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_choice(cls, option: str):
|
||||
"""Instantiate from choice."""
|
||||
@@ -28,7 +28,8 @@ class ModelOutput(BaseModel):
|
||||
raise ValidationError()
|
||||
assert isinstance(option, str)
|
||||
allowed_options_list = cls.list_fields()
|
||||
assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
|
||||
assert option in allowed_options_list, \
|
||||
f"{option} is not among allowed fields {allowed_options_list}"
|
||||
kwargs = {field: 0 for field in cls.list_fields()}
|
||||
kwargs[option] = 1
|
||||
return cls(**kwargs)
|
||||
@@ -36,15 +37,19 @@ class ModelOutput(BaseModel):
|
||||
def __repr__(self) -> str:
|
||||
model_dict = self.model_dump()
|
||||
model_repr_str = f"{self.classname()}("
|
||||
model_repr_str += ", ".join([f"{field}={value:.3f}" for field, value in model_dict.items()])
|
||||
model_repr_str += ", ".join([
|
||||
f"{field}={value:.3f}"
|
||||
for field, value
|
||||
in model_dict.items()
|
||||
])
|
||||
model_repr_str += ")"
|
||||
return model_repr_str
|
||||
|
||||
|
||||
def highest_score_field(self) -> str:
|
||||
"""Return name of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
return max(model_dict, key=lambda k: model_dict[k])
|
||||
|
||||
|
||||
def highest_score_value(self) -> float:
|
||||
"""Return value of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
|
||||
@@ -2,4 +2,4 @@ from .classes import (
|
||||
InformationValueModelOutput,
|
||||
FramingModelOutput,
|
||||
SalienceModelOutput
|
||||
)
|
||||
)
|
||||
|
||||
@@ -14,13 +14,13 @@ class ModelOutput(BaseModel):
|
||||
def list_fields(cls) -> List[str]:
|
||||
"""List options that are stored as attributes."""
|
||||
return list(cls.model_fields.keys())
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_random(cls):
|
||||
"""Instantiate with random numbers."""
|
||||
kwargs = {field: random.random() for field in cls.list_fields()}
|
||||
return cls(**kwargs)
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_choice(cls, option: str):
|
||||
"""Instantiate from choice."""
|
||||
@@ -28,7 +28,8 @@ class ModelOutput(BaseModel):
|
||||
raise ValidationError()
|
||||
assert isinstance(option, str)
|
||||
allowed_options_list = cls.list_fields()
|
||||
assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
|
||||
assert option in allowed_options_list, \
|
||||
f"{option} is not among allowed fields {allowed_options_list}"
|
||||
kwargs = {field: 0 for field in cls.list_fields()}
|
||||
kwargs[option] = 1
|
||||
return cls(**kwargs)
|
||||
@@ -36,15 +37,19 @@ class ModelOutput(BaseModel):
|
||||
def __repr__(self) -> str:
|
||||
model_dict = self.model_dump()
|
||||
model_repr_str = f"{self.classname()}("
|
||||
model_repr_str += ", ".join([f"{field}={value:.3f}" for field, value in model_dict.items()])
|
||||
model_repr_str += ", ".join([
|
||||
f"{field}={value:.3f}"
|
||||
for field, value
|
||||
in model_dict.items()
|
||||
])
|
||||
model_repr_str += ")"
|
||||
return model_repr_str
|
||||
|
||||
|
||||
def highest_score_field(self) -> str:
|
||||
"""Return name of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
return max(model_dict, key=lambda k: model_dict[k])
|
||||
|
||||
|
||||
def highest_score_value(self) -> float:
|
||||
"""Return value of field with highest score."""
|
||||
model_dict = self.model_dump()
|
||||
@@ -84,4 +89,4 @@ if __name__ == '__main__':
|
||||
m = SalienceModelOutput.from_random()
|
||||
print(repr(m))
|
||||
print(m.highest_score_field())
|
||||
print(m.highest_score_value())
|
||||
print(m.highest_score_value())
|
||||
|
||||
+7
-5
@@ -30,6 +30,7 @@ BasicAuth(app, AUTH_DICT)
|
||||
# connect to database
|
||||
collection, db, client = connect()
|
||||
|
||||
|
||||
# define callbacks
|
||||
@app.callback(
|
||||
Output("alert-element", "is_open"),
|
||||
@@ -44,6 +45,7 @@ def show_alert(
|
||||
logging.info(f"updated alert message: {msg}")
|
||||
return True, msg
|
||||
|
||||
|
||||
@app.callback(
|
||||
Output("alert-message", "data"),
|
||||
Output("vis-com-name", "data"),
|
||||
@@ -73,7 +75,7 @@ def cycle_visual_communication_data(
|
||||
annotation_values
|
||||
]
|
||||
# check if next-button clicked
|
||||
if n_clicks == 0:
|
||||
if n_clicks == 0:
|
||||
logging.info("stopping early: next-button has not yet been clicked")
|
||||
return response
|
||||
# check if visual communication name is set
|
||||
@@ -102,13 +104,13 @@ def cycle_visual_communication_data(
|
||||
in annotation_values
|
||||
]
|
||||
annotations = {
|
||||
key: value
|
||||
for key, value
|
||||
key: value
|
||||
for key, value
|
||||
in zip(annotation_keys, annotation_values)
|
||||
}
|
||||
# instantiate ModelOutputs object
|
||||
annotations = ModelOutputs.from_annotations(**annotations)
|
||||
# save data to
|
||||
# save data to database
|
||||
upsert_annotations(
|
||||
collection=collection,
|
||||
vis_com_name=vis_com_name,
|
||||
@@ -137,7 +139,7 @@ def cycle_visual_communication_data(
|
||||
# reset annotations
|
||||
annotation_values = [None for elem in annotation_values]
|
||||
except NoDocumentFoundException:
|
||||
msg = f"no unannotated data in database"
|
||||
msg = "no unannotated data in database"
|
||||
logging.warning(msg)
|
||||
response[0] = msg
|
||||
return tuple(response)
|
||||
|
||||
@@ -1 +1 @@
|
||||
from .layout import app_layout
|
||||
from .layout import app_layout
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from dash import html, dcc
|
||||
from dash import html
|
||||
import dash_bootstrap_components as dbc
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import dash_mantine_components as dmc
|
||||
from dash import dcc, html
|
||||
from typing import List
|
||||
|
||||
from .image import image_element
|
||||
from .inputs import inputs_element
|
||||
@@ -17,4 +15,4 @@ body_element = dmc.Container(
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
@@ -20,4 +20,4 @@ inputs_element = dmc.SimpleGrid(
|
||||
labels_element,
|
||||
next_button
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
@@ -20,6 +20,7 @@ from src.model_textual import (
|
||||
SalienceModelOutput
|
||||
)
|
||||
|
||||
|
||||
def generate_option_labels(model) -> List[str]:
|
||||
"""Generate presentable list of attributes from an OutputModel."""
|
||||
labels = [
|
||||
@@ -28,35 +29,51 @@ def generate_option_labels(model) -> List[str]:
|
||||
]
|
||||
return labels
|
||||
|
||||
|
||||
def generate_visual_syntax_options_map():
|
||||
"""Generate map of titles and options for visual syntax labels."""
|
||||
options_map = {}
|
||||
# add experiential labels
|
||||
options_map["visual syntax"] = generate_option_labels(VisualSyntaxModelOutput)
|
||||
options_map["visual syntax"] = generate_option_labels(
|
||||
VisualSyntaxModelOutput
|
||||
)
|
||||
return options_map
|
||||
|
||||
|
||||
def generate_interpersonal_options_map():
|
||||
"""Generate map of titles and options for interpersonal labels."""
|
||||
options_map = {}
|
||||
# add interpersonal labels
|
||||
options_map["contact"] = generate_option_labels(ContactModelOutput)
|
||||
options_map["angle"] = generate_option_labels(AngleModelOutput)
|
||||
options_map["point of view"] = generate_option_labels(PointOfViewModelOutput)
|
||||
options_map["point of view"] = generate_option_labels(
|
||||
PointOfViewModelOutput
|
||||
)
|
||||
options_map["distance"] = generate_option_labels(DistanceModelOutput)
|
||||
options_map["modality lighting"] = generate_option_labels(ModalityLightingModelOutput)
|
||||
options_map["modality color"] = generate_option_labels(ModalityColorModelOutput)
|
||||
options_map["modality depth"] = generate_option_labels(ModalityDepthModelOutput)
|
||||
options_map["modality lighting"] = generate_option_labels(
|
||||
ModalityLightingModelOutput
|
||||
)
|
||||
options_map["modality color"] = generate_option_labels(
|
||||
ModalityColorModelOutput
|
||||
)
|
||||
options_map["modality depth"] = generate_option_labels(
|
||||
ModalityDepthModelOutput
|
||||
)
|
||||
return options_map
|
||||
|
||||
|
||||
def generate_textual_options_map():
|
||||
"""Generate map of titles and options for textual labels."""
|
||||
options_map = {}
|
||||
# add textual labels
|
||||
options_map["information value"] = generate_option_labels(InformationValueModelOutput)
|
||||
options_map["information value"] = generate_option_labels(
|
||||
InformationValueModelOutput
|
||||
)
|
||||
options_map["framing"] = generate_option_labels(FramingModelOutput)
|
||||
options_map["salience"] = generate_option_labels(SalienceModelOutput)
|
||||
return options_map
|
||||
|
||||
|
||||
# prepare experiential container
|
||||
experiential_map = generate_visual_syntax_options_map()
|
||||
experiential_container = dmc.Col(
|
||||
@@ -118,4 +135,4 @@ labels_element = dmc.Grid(
|
||||
interpersonal_container,
|
||||
textual_container,
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
from dash import dcc
|
||||
import dash_mantine_components as dmc
|
||||
|
||||
from .stores import stores_element
|
||||
|
||||
@@ -5,12 +5,14 @@ import os
|
||||
storage_type = "session"
|
||||
if "ENV" in os.environ and os.getenv("ENV") == "DEV":
|
||||
storage_type = "memory"
|
||||
logging.info(f"ENV=DEV -> dcc.Stores changed to storage_type={storage_type}")
|
||||
|
||||
logging.info(
|
||||
"ENV=DEV -> dcc.Stores changed to storage_type=%s",
|
||||
storage_type
|
||||
)
|
||||
|
||||
stores_element = html.Div(
|
||||
children=[
|
||||
dcc.Store(id="alert-message", storage_type=storage_type, data=""),
|
||||
dcc.Store(id="vis-com-name", storage_type=storage_type, data=""),
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user