add_gitea_workflows #17

Merged
brian merged 48 commits from add_gitea_workflows into main 2024-02-25 18:45:08 +01:00
12 changed files with 230 additions and 215 deletions
Showing only changes of commit ce063822b5 - Show all commits
+40 -42
View File
@@ -1,29 +1,27 @@
from __future__ import annotations
from pydantic import BaseModel, field_validator, field_serializer
from PIL import Image
import logging
from base64 import b64encode
from io import BytesIO
from pathlib import Path
from base64 import b64encode
import logging
from typing import List
from src.model_experiential import (
VisualSyntaxModelOutput
)
from src.model_interpersonal import (
ContactModelOutput,
AngleModelOutput,
PointOfViewModelOutput,
DistanceModelOutput,
ModalityLightingModelOutput,
ModalityColorModelOutput,
ModalityDepthModelOutput
)
from src.model_textual import (
InformationValueModelOutput,
FramingModelOutput,
SalienceModelOutput
from model_experiential import (
VisualSyntaxModelOutput,
)
from model_interpersonal import AngleModelOutput
from model_interpersonal import ContactModelOutput
from model_interpersonal import DistanceModelOutput
from model_interpersonal import ModalityColorModelOutput
from model_interpersonal import ModalityDepthModelOutput
from model_interpersonal import ModalityLightingModelOutput
from model_interpersonal import PointOfViewModelOutput
from model_textual import FramingModelOutput
from model_textual import InformationValueModelOutput
from model_textual import SalienceModelOutput
from PIL import Image
from pydantic import BaseModel
from pydantic import field_serializer
from pydantic import field_validator
class NoDocumentFoundException(Exception):
@@ -44,7 +42,7 @@ class ModelOutputs(BaseModel):
salience: SalienceModelOutput
@classmethod
def list_fields(cls) -> List[str]:
def list_fields(cls) -> list[str]:
"""List options that are stored as attributes."""
return list(cls.model_fields.keys())
@@ -71,32 +69,32 @@ class ModelOutputs(BaseModel):
modality_depth: str,
information_value: str,
framing: str,
salience: str
salience: str,
) -> ModelOutputs:
"""Instantiate from annotation."""
kwargs = {
"visual_syntax": VisualSyntaxModelOutput
'visual_syntax': VisualSyntaxModelOutput
.from_choice(visual_syntax),
"contact": ContactModelOutput
'contact': ContactModelOutput
.from_choice(contact),
"angle": AngleModelOutput
'angle': AngleModelOutput
.from_choice(angle),
"point_of_view": PointOfViewModelOutput
'point_of_view': PointOfViewModelOutput
.from_choice(point_of_view),
"distance": DistanceModelOutput
'distance': DistanceModelOutput
.from_choice(distance),
"modality_lighting": ModalityLightingModelOutput
'modality_lighting': ModalityLightingModelOutput
.from_choice(modality_lighting),
"modality_color": ModalityColorModelOutput
'modality_color': ModalityColorModelOutput
.from_choice(modality_color),
"modality_depth": ModalityDepthModelOutput
'modality_depth': ModalityDepthModelOutput
.from_choice(modality_depth),
"information_value": InformationValueModelOutput
'information_value': InformationValueModelOutput
.from_choice(information_value),
"framing": FramingModelOutput
'framing': FramingModelOutput
.from_choice(framing),
"salience": SalienceModelOutput
.from_choice(salience)
'salience': SalienceModelOutput
.from_choice(salience),
}
return cls(**kwargs)
@@ -123,17 +121,17 @@ class VisualCommunication(BaseModel):
image.load()
return VisualCommunication(name=name, image=image)
@field_serializer("image")
@field_serializer('image')
def serialize_image(image: Image.Image) -> bytes: # type: ignore
buffer = BytesIO()
image.save(buffer, format="JPEG")
image.save(buffer, format='JPEG')
return buffer.getvalue()
@field_validator("image", mode="before")
@field_validator('image', mode='before')
@classmethod
def convert_to_image(
cls,
image: Image.Image | BytesIO | bytes
image: Image.Image | BytesIO | bytes,
) -> Image.Image:
if isinstance(image, bytes):
image = BytesIO(image)
@@ -148,12 +146,12 @@ class VisualCommunication(BaseModel):
"""Convert image to be displayed on webpage."""
# convert images to bytes string
buffer = BytesIO()
self.image.save(buffer, format="png")
img_enc = b64encode(buffer.getvalue()).decode("utf-8")
self.image.save(buffer, format='png')
img_enc = b64encode(buffer.getvalue()).decode('utf-8')
return f"data:image/png;base64, {img_enc}"
def generate_random_prediction(self, force: bool = False) -> None:
"""Generate random prediction values."""
if not force and self.prediction is not None:
logging.warning("set force=True to overwrite existing values.")
logging.warning('set force=True to overwrite existing values.')
self.prediction = ModelOutputs.from_random()
+19 -17
View File
@@ -1,23 +1,25 @@
import logging
from dotenv import load_dotenv
from pathlib import Path
import os
from __future__ import annotations
from src.web import app
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
from web import app
# prepare optional local setup
env_path = Path(__file__).parent.parent / "local.env"
env_path = Path(__file__).parent.parent / 'local.env'
load_dotenv(env_path)
# ensure env vars set
necesasary_var_list = {
"MONGO_HOST",
"MONGO_DB",
"MONGO_COLLECTION",
"MONGO_USER",
"MONGO_PASSWORD",
"DASH_AUTH_USERNAME",
"DASH_AUTH_PASSWORD"
'MONGO_HOST',
'MONGO_DB',
'MONGO_COLLECTION',
'MONGO_USER',
'MONGO_PASSWORD',
'DASH_AUTH_USERNAME',
'DASH_AUTH_PASSWORD',
}
for env_var in necesasary_var_list:
# ensure env var set
@@ -38,12 +40,12 @@ fmt = (
datefmt = '%Y-%m-%d %H:%M:%S'
logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
logging.info("initialized app")
logging.info('initialized app')
server = app.server
if __name__ == "__main__":
if __name__ == '__main__':
# prepare local env vars
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# run app
app.run(debug=True)
logging.info("started app")
logging.info('started app')
+46 -42
View File
@@ -1,29 +1,33 @@
from dash import Dash, Input, Output, State, ALL
import dash_bootstrap_components as dbc
from dash_auth import BasicAuth
from __future__ import annotations
import logging
from typing import List
from pydantic import ValidationError
import os
import dash_bootstrap_components as dbc
from dash import ALL
from dash import Dash
from dash import Input
from dash import Output
from dash import State
from dash_auth import BasicAuth
from database import connect
from database import get_visual_communication
from database import ModelOutputs
from database import NoDocumentFoundException
from database import upsert_annotations
from pydantic import ValidationError
from .layout import app_layout
from src.database import (
connect,
get_visual_communication,
NoDocumentFoundException,
upsert_annotations,
ModelOutputs
)
# setup app
app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
app.title = "visual critical discourse analysis".title()
app.title = 'visual critical discourse analysis'.title()
app.layout = app_layout
server = app.server
# setup authentication
AUTH_DICT = {
os.getenv("DASH_AUTH_USERNAME"): os.getenv("DASH_AUTH_PASSWORD")
os.getenv('DASH_AUTH_USERNAME'): os.getenv('DASH_AUTH_PASSWORD'),
}
BasicAuth(app, AUTH_DICT)
@@ -33,58 +37,58 @@ collection, db, client = connect()
# define callbacks
@app.callback(
Output("alert-element", "is_open"),
Output("alert-element", "children"),
Input("alert-message", "data")
Output('alert-element', 'is_open'),
Output('alert-element', 'children'),
Input('alert-message', 'data'),
)
def show_alert(
msg: str | None
msg: str | None,
):
if msg is None or msg == "":
return False, ""
if msg is None or msg == '':
return False, ''
logging.info(f"updated alert message: {msg}")
return True, msg
@app.callback(
Output("alert-message", "data"),
Output("vis-com-name", "data"),
Output("image-container", "src"),
Output({"type": "annotation", "index": ALL}, "value"),
Input("next-button", "n_clicks"),
State("vis-com-name", "data"),
State("image-container", "src"),
State({"type": "annotation", "index": ALL}, "id"),
State({"type": "annotation", "index": ALL}, "value"),
Output('alert-message', 'data'),
Output('vis-com-name', 'data'),
Output('image-container', 'src'),
Output({'type': 'annotation', 'index': ALL}, 'value'),
Input('next-button', 'n_clicks'),
State('vis-com-name', 'data'),
State('image-container', 'src'),
State({'type': 'annotation', 'index': ALL}, 'id'),
State({'type': 'annotation', 'index': ALL}, 'value'),
prevent_initial_call=True,
)
def cycle_visual_communication_data(
n_clicks: int,
vis_com_name: str,
image_src: str,
annotation_keys: List,
annotation_values: List,
annotation_keys: list,
annotation_values: list,
):
logging.info("began cycling visual communication data")
logging.info('began cycling visual communication data')
global collection
# prepare default response
response = [
"",
'',
vis_com_name,
image_src,
annotation_values
annotation_values,
]
# check if next-button clicked
if n_clicks == 0:
logging.info("stopping early: next-button has not yet been clicked")
logging.info('stopping early: next-button has not yet been clicked')
return response
# check if visual communication name is set
if len(vis_com_name) > 0:
logging.info("saving annotations to database: %s", vis_com_name)
logging.info('saving annotations to database: %s', vis_com_name)
try:
# extract option keys
annotation_keys = [
elem["index"]
elem['index']
for elem in annotation_keys
]
# ensure all options are set
@@ -114,7 +118,7 @@ def cycle_visual_communication_data(
upsert_annotations(
collection=collection,
vis_com_name=vis_com_name,
annotations=annotations
annotations=annotations,
)
except (ValueError, ValidationError) as exc:
msg = f"failed saving annotation: {exc}"
@@ -122,12 +126,12 @@ def cycle_visual_communication_data(
response[0] = msg
return tuple(response)
# get new visual communication
logging.info("trying to get new visual communication")
logging.info('trying to get new visual communication')
try:
# get data
vis_com = get_visual_communication(
collection=collection,
with_annotation=False
with_annotation=False,
)
# set variables
vis_com_name = vis_com.name
@@ -139,7 +143,7 @@ def cycle_visual_communication_data(
# reset annotations
annotation_values = [None for elem in annotation_values]
except NoDocumentFoundException:
msg = "no unannotated data in database"
msg = 'no unannotated data in database'
logging.warning(msg)
response[0] = msg
return tuple(response)
@@ -147,5 +151,5 @@ def cycle_visual_communication_data(
response[1] = vis_com_name
response[2] = image_src
response[3] = annotation_values
logging.info("finished getting visual communication: %s", vis_com_name)
logging.info('finished getting visual communication: %s', vis_com_name)
return tuple(response)
+47 -50
View File
@@ -1,27 +1,24 @@
from dash import html, dcc
from __future__ import annotations
import dash_mantine_components as dmc
from typing import List
from src.model_experiential import (
VisualSyntaxModelOutput
)
from src.model_interpersonal import (
ContactModelOutput,
AngleModelOutput,
PointOfViewModelOutput,
DistanceModelOutput,
ModalityLightingModelOutput,
ModalityColorModelOutput,
ModalityDepthModelOutput
)
from src.model_textual import (
InformationValueModelOutput,
FramingModelOutput,
SalienceModelOutput
from dash import dcc
from dash import html
from model_experiential import (
VisualSyntaxModelOutput,
)
from model_interpersonal import AngleModelOutput
from model_interpersonal import ContactModelOutput
from model_interpersonal import DistanceModelOutput
from model_interpersonal import ModalityColorModelOutput
from model_interpersonal import ModalityDepthModelOutput
from model_interpersonal import ModalityLightingModelOutput
from model_interpersonal import PointOfViewModelOutput
from model_textual import FramingModelOutput
from model_textual import InformationValueModelOutput
from model_textual import SalienceModelOutput
def generate_option_labels(model) -> List[str]:
def generate_option_labels(model) -> list[str]:
"""Generate presentable list of attributes from an OutputModel."""
labels = [
label.replace('_', ' ').title()
@@ -34,8 +31,8 @@ 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
@@ -44,20 +41,20 @@ 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['contact'] = generate_option_labels(ContactModelOutput)
options_map['angle'] = generate_option_labels(AngleModelOutput)
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['distance'] = generate_option_labels(DistanceModelOutput)
options_map['modality lighting'] = generate_option_labels(
ModalityLightingModelOutput,
)
options_map["modality color"] = generate_option_labels(
ModalityColorModelOutput
options_map['modality color'] = generate_option_labels(
ModalityColorModelOutput,
)
options_map["modality depth"] = generate_option_labels(
ModalityDepthModelOutput
options_map['modality depth'] = generate_option_labels(
ModalityDepthModelOutput,
)
return options_map
@@ -66,11 +63,11 @@ 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)
options_map['framing'] = generate_option_labels(FramingModelOutput)
options_map['salience'] = generate_option_labels(SalienceModelOutput)
return options_map
@@ -78,11 +75,11 @@ def generate_textual_options_map():
experiential_map = generate_visual_syntax_options_map()
experiential_container = dmc.Col(
children=[
html.H4("experiential".title()),
], span=5
html.H4('experiential'.title()),
], span=5,
)
for title, options in experiential_map.items():
id_dict = {"type": "annotation", "index": title.replace('_', '-')}
id_dict = {'type': 'annotation', 'index': title.replace('_', '-')}
experiential_container.children.append(
dmc.Container([
html.B(title.title()),
@@ -90,17 +87,17 @@ for title, options in experiential_map.items():
options=options,
id=id_dict,
),
])
]),
)
# prepare interpersonal container
interpersonal_map = generate_interpersonal_options_map()
interpersonal_container = dmc.Col(
children=[
html.H4("interpersonal".title()),
], span=3
html.H4('interpersonal'.title()),
], span=3,
)
for title, options in interpersonal_map.items():
id_dict = {"type": "annotation", "index": title.replace('_', '-')}
id_dict = {'type': 'annotation', 'index': title.replace('_', '-')}
interpersonal_container.children.append(
dmc.Container([
html.B(title.title()),
@@ -108,17 +105,17 @@ for title, options in interpersonal_map.items():
options=options,
id=id_dict,
),
])
]),
)
# prepare textual container
textual_map = generate_textual_options_map()
textual_container = dmc.Col(
children=[
html.H4("textual".title()),
], span=4
html.H4('textual'.title()),
], span=4,
)
for title, options in textual_map.items():
id_dict = {"type": "annotation", "index": title.replace('_', '-')}
id_dict = {'type': 'annotation', 'index': title.replace('_', '-')}
textual_container.children.append(
dmc.Container([
html.B(title),
@@ -126,7 +123,7 @@ for title, options in textual_map.items():
options=options,
id=id_dict,
),
])
]),
)
labels_element = dmc.Grid(
@@ -134,5 +131,5 @@ labels_element = dmc.Grid(
experiential_container,
interpersonal_container,
textual_container,
]
],
)
@@ -1,13 +1,15 @@
from __future__ import annotations
from pathlib import Path
from src.database import VisualCommunication
from database import VisualCommunication
if __name__ == "__main__":
if __name__ == '__main__':
# get list of image paths
test_dir = Path(__file__).parent
img_dir = test_dir / "imgs"
img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()]
img_dir = test_dir / 'imgs'
img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
print(img_path_list)
# instantiate data object
vis_com_list = [
@@ -16,6 +18,7 @@ if __name__ == "__main__":
in img_path_list
]
# generate random predictions
[vis_com.generate_random_prediction() for vis_com in vis_com_list]
for vis_com in vis_com_list:
vis_com.generate_random_prediction()
for vis_com in vis_com_list:
print(vis_com)
@@ -1,18 +1,18 @@
from pathlib import Path
from dotenv import load_dotenv
from __future__ import annotations
import os
from pathlib import Path
from src.database import (
connect,
get_visual_communication
)
from database import connect
from database import get_visual_communication
from dotenv import load_dotenv
if __name__ == "__main__":
if __name__ == '__main__':
# prepare env vars
env_path = Path(__file__).parent.parent / "local.env"
env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
print(client.server_info())
@@ -1,15 +1,18 @@
from pathlib import Path
from dotenv import load_dotenv
from __future__ import annotations
import os
from pathlib import Path
from src.database import VisualCommunication, connect
from database import connect
from database import VisualCommunication
from dotenv import load_dotenv
if __name__ == "__main__":
if __name__ == '__main__':
# prepare env vars
env_path = Path(__file__).parent.parent / "local.env"
env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
print(client.server_info())
@@ -17,7 +20,7 @@ if __name__ == "__main__":
data = None
for data in collection.find().limit(3):
if data is None:
print("no document found")
print('no document found')
break
vis_com: VisualCommunication = VisualCommunication.model_validate(data)
print(repr(vis_com))
@@ -1,15 +1,18 @@
from __future__ import annotations
import os
from pathlib import Path
from database import connect
from database import VisualCommunication
from dotenv import load_dotenv
from pymongo.errors import DuplicateKeyError
import os
from src.database import VisualCommunication, connect
if __name__ == "__main__":
if __name__ == '__main__':
# get list of image paths
test_dir = Path(__file__).parent
img_dir = test_dir / "imgs"
img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()]
img_dir = test_dir / 'imgs'
img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
print(img_path_list)
# instantiate data object
vis_com_list = [
@@ -20,10 +23,10 @@ if __name__ == "__main__":
for vis_com in vis_com_list:
print(repr(vis_com))
# prepare env vars
env_path = test_dir.parent / "local.env"
env_path = test_dir.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
print(client.server_info())
@@ -32,6 +35,6 @@ if __name__ == "__main__":
try:
result = collection.insert_one(vis_com.model_dump())
except DuplicateKeyError as exc:
print("ignoring:\n", exc)
print('ignoring:\n', exc)
else:
print(f"inserted document: {result}")
@@ -1,7 +1,9 @@
from src.database import ModelOutputs
from __future__ import annotations
from database import ModelOutputs
if __name__ == "__main__":
if __name__ == '__main__':
# instantiate data object
vis_com_list = [
ModelOutputs.from_random()
@@ -1,15 +1,15 @@
from pathlib import Path
from dotenv import load_dotenv
import os
from __future__ import annotations
import logging
import os
from pathlib import Path
from src.database import (
VisualCommunication,
connect,
upsert_predictions
)
from database import connect
from database import upsert_predictions
from database import VisualCommunication
from dotenv import load_dotenv
if __name__ == "__main__":
if __name__ == '__main__':
# setup logging
fmt = (
'%(asctime)s | '
@@ -22,8 +22,8 @@ if __name__ == "__main__":
logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
# get list of image paths
test_dir = Path(__file__).parent
img_dir = test_dir / "imgs"
img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()]
img_dir = test_dir / 'imgs'
img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
# instantiate data object
vis_com_list = [
VisualCommunication.from_file(path)
@@ -31,16 +31,19 @@ if __name__ == "__main__":
in img_path_list
]
# generate random predictions
[vis_com.generate_random_prediction() for vis_com in vis_com_list]
for vis_com in vis_com_list:
vis_com.generate_random_prediction()
# prepare env vars
env_path = test_dir.parent / "local.env"
env_path = test_dir.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
# upload visual communication
for vis_com in vis_com_list:
if vis_com.prediction is None:
continue
upsert_predictions(
collection=collection,
vis_com_name=vis_com.name,
@@ -1,18 +1,18 @@
from pathlib import Path
from dotenv import load_dotenv
from __future__ import annotations
import os
from pathlib import Path
from src.database import (
connect,
total_annotated
)
from database import connect
from database import total_annotated
from dotenv import load_dotenv
if __name__ == "__main__":
if __name__ == '__main__':
# prepare env vars
env_path = Path(__file__).parent.parent / "local.env"
env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
# get visual communication
@@ -1,18 +1,18 @@
from pathlib import Path
from dotenv import load_dotenv
from __future__ import annotations
import os
from pathlib import Path
from src.database import (
connect,
total_documents
)
from database import connect
from database import total_documents
from dotenv import load_dotenv
if __name__ == "__main__":
if __name__ == '__main__':
# prepare env vars
env_path = Path(__file__).parent.parent / "local.env"
env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
# get visual communication