brian/visual_critical_discourse_analysis
1
0
Fork 0
Code Issues 11 Pull Requests Actions Projects 1 Releases Activity

Merge pull request 'add_gitea_workflows' (#17) from add_gitea_workflows into main
Code Quality Pipeline / Test (push) Successful in 52s
Details
pipeline / Show (push) Successful in 6s
Details

Browse Source 
Reviewed-on: http://192.168.1.2:3000/brian/visual_critical_discourse_analysis/pulls/17
This commit was merged in pull request #17.
This commit is contained in:
Brian Bjarke Jensen
2024-02-25 18:45:07 +01:00
parent 113ac2146f 7334b27849
commit a8459d356d
37 changed files with 667 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.flake8
+2
View File
@@ -0,0 +1,2 @@
[flake8]
per-file-ignores = __init__.py:F401
.gitea/workflows/default.yaml
+27
View File
@@ -0,0 +1,27 @@
name: Code Quality Pipeline
run-name: ${{ gitea.actor }} is running the Code Quality Pipeline
runs-on: ubuntu-latest
on: push
image: python:3.12
jobs:
test:
name: Test
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
- name: Setup Environment
uses: https://github.com/actions/setup-python@v3
with:
python-verison: "3.12"
architecture: "x64"
- name: Install Packages
run: |
pip install poetry
poetry install
- name: PEP8 Check
run: |
poetry run flake8 ./src --benchmark
- name: Type Check
run: |
poetry run mypy ./src --disable-error-code=import-untyped
.gitea/workflows/publish.yaml
+52
View File
@@ -0,0 +1,52 @@
name: CI Pipeline
run-name: ${{ gitea.actor }} is running the CI Pipeline
runs-on: ubuntu-latest
on:
pull_request:
branches:
- main
jobs:
test:
name: Test
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
- name: Setup Environment
uses: https://github.com/actions/setup-python@v3
with:
python-verison: "3.12"
architecture: "x64"
- name: Install Packages
run: |
pip install poetry
poetry install
- name: PEP8 Check
run: |
poetry run flake8 ./src --benchmark
- name: Type Check
run: |
poetry run mypy ./src --disable-error-code=import-untyped
publish:
name: Build and Publish
runs-on: ubuntu-latest
needs: [test]
steps:
- name: Checkout Code
uses: actions/checkout@v3
- name: Set Environment Variables
run: |
echo "sha_short=$(git rev-parse --short ${{ gitea.sha }} )" >> "$GITHUB_ENV"
- name: Show Environment Variables
run: |
echo "DOCKER_REPO_URL: ${{ vars.docker_repo_url }}"
echo "REPOSITORY: ${{ gitea.repository }}"
echo "COMMIT_SHA: ${{ env.sha_short }}"
- name: Build and Push Image
uses: docker/build-push-action@v2
with:
context: .
push: true
tags: |
${{ vars.docker_repo_url }}/${{ gitea.repository }}:${{ env.sha_short }}
${{ vars.docker_repo_url }}/${{ gitea.repository }}:latest
.gitea/workflows/show_env_vars.yaml
+24
View File
@@ -0,0 +1,24 @@
name: pipeline
run-name: ${{ gitea.actor }} is running the script
runs-on: ubuntu-latest
on: push
image: python:3.12
jobs:
show:
name: Show
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
- name: Set Environment Variables
run: |
echo "REPOSITORY: ${{ gitea.repository }}"
echo "DOCKER_REPO_URL: ${{ vars.docker_repo_url }}"
echo "COMMIT_SHA: ${{ gitea.sha }}"
echo $(git rev-parse --short ${{ gitea.sha }})
echo "sha_short=$(git rev-parse --short ${{ gitea.sha }} )" >> "$GITHUB_ENV"
- name: Show Environment Variables
run: |
echo "REPOSITORY: ${{ gitea.repository }}"
echo "DOCKER_REPO_URL: ${{ vars.docker_repo_url }}"
echo "COMMIT_SHA: ${{ env.sha_short }}"
.pre-commit-config.yaml
+44
View File
@@ -0,0 +1,44 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: debug-statements
- id: double-quote-string-fixer
- id: name-tests-test
- id: requirements-txt-fixer
- repo: https://github.com/asottile/setup-cfg-fmt
rev: v2.5.0
hooks:
- id: setup-cfg-fmt
- repo: https://github.com/asottile/reorder-python-imports
rev: v3.12.0
hooks:
- id: reorder-python-imports
exclude: ^(pre_commit/resources/|testing/resources/python3_hooks_repo/)
args: [--py39-plus, --add-import, 'from __future__ import annotations']
- repo: https://github.com/asottile/add-trailing-comma
rev: v3.1.0
hooks:
- id: add-trailing-comma
- repo: https://github.com/asottile/pyupgrade
rev: v3.15.1
hooks:
- id: pyupgrade
args: [--py39-plus]
- repo: https://github.com/hhatto/autopep8
rev: v2.0.4
hooks:
- id: autopep8
- repo: https://github.com/PyCQA/flake8
rev: 7.0.0
hooks:
- id: flake8
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0
hooks:
- id: mypy
additional_dependencies: [types-all]
exclude: ^testing/resources/
poetry.lock
Generated
+118 -1
View File
@@ -287,6 +287,22 @@ idna = ["idna (>=3.6)"]
trio = ["trio (>=0.23)"] trio = ["trio (>=0.23)"]
wmi = ["wmi (>=1.5.1)"] wmi = ["wmi (>=1.5.1)"]
[[package]]
name = "flake8"
version = "7.0.0"
description = "the modular source code checker: pep8 pyflakes and co"
optional = false
python-versions = ">=3.8.1"
files = [
{file = "flake8-7.0.0-py2.py3-none-any.whl", hash = "sha256:a6dfbb75e03252917f2473ea9653f7cd799c3064e54d4c8140044c5c065f53c3"},
{file = "flake8-7.0.0.tar.gz", hash = "sha256:33f96621059e65eec474169085dc92bf26e7b2d47366b70be2f67ab80dc25132"},
]
[package.dependencies]
mccabe = ">=0.7.0,<0.8.0"
pycodestyle = ">=2.11.0,<2.12.0"
pyflakes = ">=3.2.0,<3.3.0"
[[package]] [[package]]
name = "flask" name = "flask"
version = "3.0.2" version = "3.0.2"
@@ -456,6 +472,74 @@ files = [
{file = "MarkupSafe-2.1.5.tar.gz", hash = "sha256:d283d37a890ba4c1ae73ffadf8046435c76e7bc2247bbb63c00bd1a709c6544b"}, {file = "MarkupSafe-2.1.5.tar.gz", hash = "sha256:d283d37a890ba4c1ae73ffadf8046435c76e7bc2247bbb63c00bd1a709c6544b"},
] ]
[[package]]
name = "mccabe"
version = "0.7.0"
description = "McCabe checker, plugin for flake8"
optional = false
python-versions = ">=3.6"
files = [
{file = "mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e"},
{file = "mccabe-0.7.0.tar.gz", hash = "sha256:348e0240c33b60bbdf4e523192ef919f28cb2c3d7d5c7794f74009290f236325"},
]
[[package]]
name = "mypy"
version = "1.8.0"
description = "Optional static typing for Python"
optional = false
python-versions = ">=3.8"
files = [
{file = "mypy-1.8.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:485a8942f671120f76afffff70f259e1cd0f0cfe08f81c05d8816d958d4577d3"},
{file = "mypy-1.8.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:df9824ac11deaf007443e7ed2a4a26bebff98d2bc43c6da21b2b64185da011c4"},
{file = "mypy-1.8.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2afecd6354bbfb6e0160f4e4ad9ba6e4e003b767dd80d85516e71f2e955ab50d"},
{file = "mypy-1.8.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:8963b83d53ee733a6e4196954502b33567ad07dfd74851f32be18eb932fb1cb9"},
{file = "mypy-1.8.0-cp310-cp310-win_amd64.whl", hash = "sha256:e46f44b54ebddbeedbd3d5b289a893219065ef805d95094d16a0af6630f5d410"},
{file = "mypy-1.8.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:855fe27b80375e5c5878492f0729540db47b186509c98dae341254c8f45f42ae"},
{file = "mypy-1.8.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4c886c6cce2d070bd7df4ec4a05a13ee20c0aa60cb587e8d1265b6c03cf91da3"},
{file = "mypy-1.8.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d19c413b3c07cbecf1f991e2221746b0d2a9410b59cb3f4fb9557f0365a1a817"},
{file = "mypy-1.8.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:9261ed810972061388918c83c3f5cd46079d875026ba97380f3e3978a72f503d"},
{file = "mypy-1.8.0-cp311-cp311-win_amd64.whl", hash = "sha256:51720c776d148bad2372ca21ca29256ed483aa9a4cdefefcef49006dff2a6835"},
{file = "mypy-1.8.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:52825b01f5c4c1c4eb0db253ec09c7aa17e1a7304d247c48b6f3599ef40db8bd"},
{file = "mypy-1.8.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:f5ac9a4eeb1ec0f1ccdc6f326bcdb464de5f80eb07fb38b5ddd7b0de6bc61e55"},
{file = "mypy-1.8.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:afe3fe972c645b4632c563d3f3eff1cdca2fa058f730df2b93a35e3b0c538218"},
{file = "mypy-1.8.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:42c6680d256ab35637ef88891c6bd02514ccb7e1122133ac96055ff458f93fc3"},
{file = "mypy-1.8.0-cp312-cp312-win_amd64.whl", hash = "sha256:720a5ca70e136b675af3af63db533c1c8c9181314d207568bbe79051f122669e"},
{file = "mypy-1.8.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:028cf9f2cae89e202d7b6593cd98db6759379f17a319b5faf4f9978d7084cdc6"},
{file = "mypy-1.8.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:4e6d97288757e1ddba10dd9549ac27982e3e74a49d8d0179fc14d4365c7add66"},
{file = "mypy-1.8.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7f1478736fcebb90f97e40aff11a5f253af890c845ee0c850fe80aa060a267c6"},
{file = "mypy-1.8.0-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:42419861b43e6962a649068a61f4a4839205a3ef525b858377a960b9e2de6e0d"},
{file = "mypy-1.8.0-cp38-cp38-win_amd64.whl", hash = "sha256:2b5b6c721bd4aabaadead3a5e6fa85c11c6c795e0c81a7215776ef8afc66de02"},
{file = "mypy-1.8.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:5c1538c38584029352878a0466f03a8ee7547d7bd9f641f57a0f3017a7c905b8"},
{file = "mypy-1.8.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:4ef4be7baf08a203170f29e89d79064463b7fc7a0908b9d0d5114e8009c3a259"},
{file = "mypy-1.8.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7178def594014aa6c35a8ff411cf37d682f428b3b5617ca79029d8ae72f5402b"},
{file = "mypy-1.8.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:ab3c84fa13c04aeeeabb2a7f67a25ef5d77ac9d6486ff33ded762ef353aa5592"},
{file = "mypy-1.8.0-cp39-cp39-win_amd64.whl", hash = "sha256:99b00bc72855812a60d253420d8a2eae839b0afa4938f09f4d2aa9bb4654263a"},
{file = "mypy-1.8.0-py3-none-any.whl", hash = "sha256:538fd81bb5e430cc1381a443971c0475582ff9f434c16cd46d2c66763ce85d9d"},
{file = "mypy-1.8.0.tar.gz", hash = "sha256:6ff8b244d7085a0b425b56d327b480c3b29cafbd2eff27316a004f9a7391ae07"},
]
[package.dependencies]
mypy-extensions = ">=1.0.0"
typing-extensions = ">=4.1.0"
[package.extras]
dmypy = ["psutil (>=4.0)"]
install-types = ["pip"]
mypyc = ["setuptools (>=50)"]
reports = ["lxml"]
[[package]]
name = "mypy-extensions"
version = "1.0.0"
description = "Type system extensions for programs checked with the mypy type checker."
optional = false
python-versions = ">=3.5"
files = [
{file = "mypy_extensions-1.0.0-py3-none-any.whl", hash = "sha256:4392f6c0eb8a5668a69e23d168ffa70f0be9ccfd32b5cc2d26a34ae5b844552d"},
{file = "mypy_extensions-1.0.0.tar.gz", hash = "sha256:75dbf8955dc00442a438fc4d0666508a9a97b6bd41aa2f0ffe9d2f2725af0782"},
]
[[package]] [[package]]
name = "nest-asyncio" name = "nest-asyncio"
version = "1.6.0" version = "1.6.0"
@@ -578,6 +662,17 @@ files = [
packaging = "*" packaging = "*"
tenacity = ">=6.2.0" tenacity = ">=6.2.0"
[[package]]
name = "pycodestyle"
version = "2.11.1"
description = "Python style guide checker"
optional = false
python-versions = ">=3.8"
files = [
{file = "pycodestyle-2.11.1-py2.py3-none-any.whl", hash = "sha256:44fe31000b2d866f2e41841b18528a505fbd7fef9017b04eff4e2648a0fadc67"},
{file = "pycodestyle-2.11.1.tar.gz", hash = "sha256:41ba0e7afc9752dfb53ced5489e89f8186be00e599e712660695b7a75ff2663f"},
]
[[package]] [[package]]
name = "pydantic" name = "pydantic"
version = "2.6.1" version = "2.6.1"
@@ -688,6 +783,17 @@ files = [
[package.dependencies] [package.dependencies]
typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0" typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0"
[[package]]
name = "pyflakes"
version = "3.2.0"
description = "passive checker of Python programs"
optional = false
python-versions = ">=3.8"
files = [
{file = "pyflakes-3.2.0-py2.py3-none-any.whl", hash = "sha256:84b5be138a2dfbb40689ca07e2152deb896a65c3a3e24c251c5c62489568074a"},
{file = "pyflakes-3.2.0.tar.gz", hash = "sha256:1c61603ff154621fb2a9172037d84dca3500def8c8b630657d1701f026f8af3f"},
]
[[package]] [[package]]
name = "pymongo" name = "pymongo"
version = "4.6.1" version = "4.6.1"
@@ -881,6 +987,17 @@ files = [
[package.extras] [package.extras]
doc = ["reno", "sphinx", "tornado (>=4.5)"] doc = ["reno", "sphinx", "tornado (>=4.5)"]
[[package]]
name = "types-pillow"
version = "10.2.0.20240213"
description = "Typing stubs for Pillow"
optional = false
python-versions = ">=3.8"
files = [
{file = "types-Pillow-10.2.0.20240213.tar.gz", hash = "sha256:4800b61bf7eabdae2f1b17ade0d080709ed33e9f26a2e900e470e8b56ebe2387"},
{file = "types_Pillow-10.2.0.20240213-py3-none-any.whl", hash = "sha256:062c5a0f20301a30f2df4db583f15b3c2a1283a12518d1f9d81396154e12c1af"},
]
[[package]] [[package]]
name = "typing-extensions" name = "typing-extensions"
version = "4.9.0" version = "4.9.0"
@@ -944,4 +1061,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
[metadata] [metadata]
lock-version = "2.0" lock-version = "2.0"
python-versions = "^3.12" python-versions = "^3.12"
content-hash = "d4df77270fa7781e482a10c4671c5c22d23f4d461bbe007f98b929142078a653" content-hash = "659fbec7604539854b960d7ffaf5c61ab80ae9fba4f6ab320a615a3104710122"
pyproject.toml
+5
View File
@@ -21,6 +21,11 @@ pymongo = "^4.6.1"
dash-auth = "^2.2.0" dash-auth = "^2.2.0"
[tool.poetry.group.test.dependencies]
flake8 = "^7.0.0"
mypy = "^1.8.0"
types-pillow = "^10.2.0.20240213"
[build-system] [build-system]
requires = ["poetry-core"] requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api" build-backend = "poetry.core.masonry.api"
src/__init__.py
View File
src/database/classes.py
+62 -47
View File
@@ -1,29 +1,33 @@
from __future__ import annotations 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 io import BytesIO
from pathlib import Path from pathlib import Path
from base64 import b64encode
import logging from PIL import Image
from typing import List from pydantic import BaseModel
from pydantic import field_serializer
from pydantic import field_validator
from src.model_experiential import ( from src.model_experiential import (
VisualSyntaxModelOutput VisualSyntaxModelOutput,
)
from src.model_interpersonal import (
ContactModelOutput,
AngleModelOutput,
PointOfViewModelOutput,
DistanceModelOutput,
ModalityLightingModelOutput,
ModalityColorModelOutput,
ModalityDepthModelOutput
)
from src.model_textual import (
InformationValueModelOutput,
FramingModelOutput,
SalienceModelOutput
) )
from src.model_interpersonal import AngleModelOutput
from src.model_interpersonal import ContactModelOutput
from src.model_interpersonal import DistanceModelOutput
from src.model_interpersonal import ModalityColorModelOutput
from src.model_interpersonal import ModalityDepthModelOutput
from src.model_interpersonal import ModalityLightingModelOutput
from src.model_interpersonal import PointOfViewModelOutput
from src.model_textual import FramingModelOutput
from src.model_textual import InformationValueModelOutput
from src.model_textual import SalienceModelOutput
class NoDocumentFoundException(Exception):
pass
class ModelOutputs(BaseModel): class ModelOutputs(BaseModel):
visual_syntax: VisualSyntaxModelOutput visual_syntax: VisualSyntaxModelOutput
@@ -39,7 +43,7 @@ class ModelOutputs(BaseModel):
salience: SalienceModelOutput salience: SalienceModelOutput
@classmethod @classmethod
def list_fields(cls) -> List[str]: def list_fields(cls) -> list[str]:
"""List options that are stored as attributes.""" """List options that are stored as attributes."""
return list(cls.model_fields.keys()) return list(cls.model_fields.keys())
@@ -47,12 +51,12 @@ class ModelOutputs(BaseModel):
def from_random(cls) -> ModelOutputs: def from_random(cls) -> ModelOutputs:
"""Instantiate with random numbers.""" """Instantiate with random numbers."""
kwargs = { kwargs = {
field: field_info.annotation.from_random() field: field_info.annotation.from_random() # type: ignore
for field, field_info for field, field_info
in cls.model_fields.items() in cls.model_fields.items()
} }
return cls(**kwargs) return cls(**kwargs)
@classmethod @classmethod
def from_annotations( def from_annotations(
cls, cls,
@@ -66,21 +70,32 @@ class ModelOutputs(BaseModel):
modality_depth: str, modality_depth: str,
information_value: str, information_value: str,
framing: str, framing: str,
salience: str salience: str,
) -> ModelOutputs: ) -> ModelOutputs:
"""Instantiate from annotation.""" """Instantiate from annotation."""
kwargs = { kwargs = {
"visual_syntax": VisualSyntaxModelOutput.from_choice(visual_syntax), 'visual_syntax': VisualSyntaxModelOutput
"contact": ContactModelOutput.from_choice(contact), .from_choice(visual_syntax),
"angle": AngleModelOutput.from_choice(angle), 'contact': ContactModelOutput
"point_of_view": PointOfViewModelOutput.from_choice(point_of_view), .from_choice(contact),
"distance": DistanceModelOutput.from_choice(distance), 'angle': AngleModelOutput
"modality_lighting": ModalityLightingModelOutput.from_choice(modality_lighting), .from_choice(angle),
"modality_color": ModalityColorModelOutput.from_choice(modality_color), 'point_of_view': PointOfViewModelOutput
"modality_depth": ModalityDepthModelOutput.from_choice(modality_depth), .from_choice(point_of_view),
"information_value": InformationValueModelOutput.from_choice(information_value), 'distance': DistanceModelOutput
"framing": FramingModelOutput.from_choice(framing), .from_choice(distance),
"salience": SalienceModelOutput.from_choice(salience) '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) return cls(**kwargs)
@@ -107,15 +122,18 @@ class VisualCommunication(BaseModel):
image.load() image.load()
return VisualCommunication(name=name, image=image) return VisualCommunication(name=name, image=image)
@field_serializer("image") @field_serializer('image')
def serialize_image(image: Image.Image) -> bytes: def serialize_image(image: Image.Image) -> bytes: # type: ignore
buffer = BytesIO() buffer = BytesIO()
image.save(buffer, format="JPEG") image.save(buffer, format='JPEG')
return buffer.getvalue() return buffer.getvalue()
@field_validator("image", mode="before") @field_validator('image', mode='before')
@classmethod @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): if isinstance(image, bytes):
image = BytesIO(image) image = BytesIO(image)
if isinstance(image, BytesIO): if isinstance(image, BytesIO):
@@ -124,20 +142,17 @@ class VisualCommunication(BaseModel):
def __repr__(self) -> str: def __repr__(self) -> str:
return f"{self.classname()}(name='{self.name}')" return f"{self.classname()}(name='{self.name}')"
def webencoded_image(self) -> str: def webencoded_image(self) -> str:
"""Convert image to be displayed on webpage.""" """Convert image to be displayed on webpage."""
# convert images to bytes string # convert images to bytes string
buffer = BytesIO() buffer = BytesIO()
self.image.save(buffer, format="png") self.image.save(buffer, format='png')
img_enc = b64encode(buffer.getvalue()).decode("utf-8") img_enc = b64encode(buffer.getvalue()).decode('utf-8')
return f"data:image/png;base64, {img_enc}" return f"data:image/png;base64, {img_enc}"
def generate_random_prediction(self, force: bool = False) -> None: def generate_random_prediction(self, force: bool = False) -> None:
"""Generate random prediction values.""" """Generate random prediction values."""
if not force and self.prediction is not None: 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() self.prediction = ModelOutputs.from_random()
class NoDocumentFoundException(Exception):
pass
src/database/utils.py
+24 -13
View File
@@ -20,32 +20,39 @@ def total_annotated(
) -> int: ) -> int:
"""Get total number of annotated documents in database.""" """Get total number of annotated documents in database."""
query = { query = {
"annotation": { "$ne": None } "annotation": {
"$ne": None
}
} }
return collection.count_documents(filter=query) return collection.count_documents(filter=query)
def get_visual_communication( def get_visual_communication(
collection: Collection, collection: Collection,
with_annotation: bool = False with_annotation: bool = False
) -> VisualCommunication: ) -> VisualCommunication:
"""Get a random visual communication from the database.""" """Get a random visual communication from the database."""
query = {} query = {}
if with_annotation: if with_annotation:
query["annotation"] = {"$ne": None} query["annotation"] = {"$ne": None}
else: else:
query["annotation"] = None query["annotation"] = {"$eq": None}
data = collection.aggregate([ 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 data_list = list(data) # read data from cursor object
if len(data) == 0: if len(data_list) == 0:
logging.error("failed getting visual communication") logging.error("failed getting visual communication")
raise NoDocumentFoundException() raise NoDocumentFoundException()
data = data[0]
logging.info("finished") logging.info("finished")
return VisualCommunication.model_validate(data) return VisualCommunication.model_validate(data_list[0])
def upsert_predictions( def upsert_predictions(
@@ -58,7 +65,9 @@ def upsert_predictions(
"name": vis_com_name "name": vis_com_name
} }
update = { update = {
"$set": { "prediction": predictions.model_dump() } "$set": {
"prediction": predictions.model_dump()
}
} }
res = collection.update_one( res = collection.update_one(
filter=query, filter=query,
@@ -79,7 +88,9 @@ def upsert_annotations(
"name": vis_com_name "name": vis_com_name
} }
update = { update = {
"$set": { "annotation": annotations.model_dump() } "$set": {
"annotation": annotations.model_dump()
}
} }
res = collection.update_one( res = collection.update_one(
filter=query, filter=query,
src/main.py
+19 -17
View File
@@ -1,22 +1,26 @@
import logging from __future__ import annotations
from dotenv import load_dotenv
from pathlib import Path
import logging import logging
import os import os
from pathlib import Path
from dotenv import load_dotenv
from src.web import app
# prepare optional local setup # prepare optional local setup
env_path = Path(__file__).parent.parent / "local.env" env_path = Path(__file__).parent.parent / 'local.env'
load_dotenv(env_path) load_dotenv(env_path)
# ensure env vars set # ensure env vars set
necesasary_var_list = { necesasary_var_list = {
"MONGO_HOST", 'MONGO_HOST',
"MONGO_DB", 'MONGO_DB',
"MONGO_COLLECTION", 'MONGO_COLLECTION',
"MONGO_USER", 'MONGO_USER',
"MONGO_PASSWORD", 'MONGO_PASSWORD',
"DASH_AUTH_USERNAME", 'DASH_AUTH_USERNAME',
"DASH_AUTH_PASSWORD" 'DASH_AUTH_PASSWORD',
} }
for env_var in necesasary_var_list: for env_var in necesasary_var_list:
# ensure env var set # ensure env var set
@@ -37,14 +41,12 @@ fmt = (
datefmt = '%Y-%m-%d %H:%M:%S' datefmt = '%Y-%m-%d %H:%M:%S'
logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO) logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
logging.info("initialized app") logging.info('initialized app')
from src.web import app
server = app.server server = app.server
if __name__ == "__main__": if __name__ == '__main__':
# prepare local env vars # prepare local env vars
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# run app # run app
app.run(debug=True) app.run(debug=True)
logging.info("started app") logging.info('started app')
src/model_experiential/__init__.py
-23
View File
@@ -1,24 +1 @@
from .classes import VisualSyntaxModelOutput 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"
# ]
src/model_experiential/classes.py
+12 -6
View File
@@ -6,6 +6,7 @@ import random
class OptionNotSetException(Exception): class OptionNotSetException(Exception):
pass pass
class ModelOutput(BaseModel): class ModelOutput(BaseModel):
@classmethod @classmethod
@@ -17,13 +18,13 @@ class ModelOutput(BaseModel):
def list_fields(cls) -> List[str]: def list_fields(cls) -> List[str]:
"""List options that are stored as attributes.""" """List options that are stored as attributes."""
return list(cls.model_fields.keys()) return list(cls.model_fields.keys())
@classmethod @classmethod
def from_random(cls): def from_random(cls):
"""Instantiate with random numbers.""" """Instantiate with random numbers."""
kwargs = {field: random.random() for field in cls.list_fields()} kwargs = {field: random.random() for field in cls.list_fields()}
return cls(**kwargs) return cls(**kwargs)
@classmethod @classmethod
def from_choice(cls, option: str): def from_choice(cls, option: str):
"""Instantiate from choice.""" """Instantiate from choice."""
@@ -31,7 +32,8 @@ class ModelOutput(BaseModel):
raise ValidationError() raise ValidationError()
assert isinstance(option, str), "option is not a string" assert isinstance(option, str), "option is not a string"
allowed_options_list = cls.list_fields() 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 = {field: 0 for field in cls.list_fields()}
kwargs[option] = 1 kwargs[option] = 1
return cls(**kwargs) return cls(**kwargs)
@@ -39,15 +41,19 @@ class ModelOutput(BaseModel):
def __repr__(self) -> str: def __repr__(self) -> str:
model_dict = self.model_dump() model_dict = self.model_dump()
model_repr_str = f"{self.classname()}(" 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 += ")" model_repr_str += ")"
return model_repr_str return model_repr_str
def highest_score_field(self) -> str: def highest_score_field(self) -> str:
"""Return name of field with highest score.""" """Return name of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
return max(model_dict, key=lambda k: model_dict[k]) return max(model_dict, key=lambda k: model_dict[k])
def highest_score_value(self) -> float: def highest_score_value(self) -> float:
"""Return value of field with highest score.""" """Return value of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
src/model_interpersonal/__init__.py
+1 -1
View File
@@ -6,4 +6,4 @@ from .classes import (
ModalityLightingModelOutput, ModalityLightingModelOutput,
ModalityColorModelOutput, ModalityColorModelOutput,
ModalityDepthModelOutput ModalityDepthModelOutput
) )
src/model_interpersonal/classes.py
+11 -6
View File
@@ -14,13 +14,13 @@ class ModelOutput(BaseModel):
def list_fields(cls) -> List[str]: def list_fields(cls) -> List[str]:
"""List options that are stored as attributes.""" """List options that are stored as attributes."""
return list(cls.model_fields.keys()) return list(cls.model_fields.keys())
@classmethod @classmethod
def from_random(cls): def from_random(cls):
"""Instantiate with random numbers.""" """Instantiate with random numbers."""
kwargs = {field: random.random() for field in cls.list_fields()} kwargs = {field: random.random() for field in cls.list_fields()}
return cls(**kwargs) return cls(**kwargs)
@classmethod @classmethod
def from_choice(cls, option: str): def from_choice(cls, option: str):
"""Instantiate from choice.""" """Instantiate from choice."""
@@ -28,7 +28,8 @@ class ModelOutput(BaseModel):
raise ValidationError() raise ValidationError()
assert isinstance(option, str) assert isinstance(option, str)
allowed_options_list = cls.list_fields() 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 = {field: 0 for field in cls.list_fields()}
kwargs[option] = 1 kwargs[option] = 1
return cls(**kwargs) return cls(**kwargs)
@@ -36,15 +37,19 @@ class ModelOutput(BaseModel):
def __repr__(self) -> str: def __repr__(self) -> str:
model_dict = self.model_dump() model_dict = self.model_dump()
model_repr_str = f"{self.classname()}(" 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 += ")" model_repr_str += ")"
return model_repr_str return model_repr_str
def highest_score_field(self) -> str: def highest_score_field(self) -> str:
"""Return name of field with highest score.""" """Return name of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
return max(model_dict, key=lambda k: model_dict[k]) return max(model_dict, key=lambda k: model_dict[k])
def highest_score_value(self) -> float: def highest_score_value(self) -> float:
"""Return value of field with highest score.""" """Return value of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
src/model_textual/__init__.py
+1 -1
View File
@@ -2,4 +2,4 @@ from .classes import (
InformationValueModelOutput, InformationValueModelOutput,
FramingModelOutput, FramingModelOutput,
SalienceModelOutput SalienceModelOutput
) )
src/model_textual/classes.py
+12 -7
View File
@@ -14,13 +14,13 @@ class ModelOutput(BaseModel):
def list_fields(cls) -> List[str]: def list_fields(cls) -> List[str]:
"""List options that are stored as attributes.""" """List options that are stored as attributes."""
return list(cls.model_fields.keys()) return list(cls.model_fields.keys())
@classmethod @classmethod
def from_random(cls): def from_random(cls):
"""Instantiate with random numbers.""" """Instantiate with random numbers."""
kwargs = {field: random.random() for field in cls.list_fields()} kwargs = {field: random.random() for field in cls.list_fields()}
return cls(**kwargs) return cls(**kwargs)
@classmethod @classmethod
def from_choice(cls, option: str): def from_choice(cls, option: str):
"""Instantiate from choice.""" """Instantiate from choice."""
@@ -28,7 +28,8 @@ class ModelOutput(BaseModel):
raise ValidationError() raise ValidationError()
assert isinstance(option, str) assert isinstance(option, str)
allowed_options_list = cls.list_fields() 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 = {field: 0 for field in cls.list_fields()}
kwargs[option] = 1 kwargs[option] = 1
return cls(**kwargs) return cls(**kwargs)
@@ -36,15 +37,19 @@ class ModelOutput(BaseModel):
def __repr__(self) -> str: def __repr__(self) -> str:
model_dict = self.model_dump() model_dict = self.model_dump()
model_repr_str = f"{self.classname()}(" 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 += ")" model_repr_str += ")"
return model_repr_str return model_repr_str
def highest_score_field(self) -> str: def highest_score_field(self) -> str:
"""Return name of field with highest score.""" """Return name of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
return max(model_dict, key=lambda k: model_dict[k]) return max(model_dict, key=lambda k: model_dict[k])
def highest_score_value(self) -> float: def highest_score_value(self) -> float:
"""Return value of field with highest score.""" """Return value of field with highest score."""
model_dict = self.model_dump() model_dict = self.model_dump()
@@ -84,4 +89,4 @@ if __name__ == '__main__':
m = SalienceModelOutput.from_random() m = SalienceModelOutput.from_random()
print(repr(m)) print(repr(m))
print(m.highest_score_field()) print(m.highest_score_field())
print(m.highest_score_value()) print(m.highest_score_value())
src/web/__init__.py
+4 -4
View File
@@ -1,4 +1,4 @@
from .app import ( from __future__ import annotations
app,
server from src.web.app import app
) from src.web.app import server
src/web/app.py
+54 -48
View File
@@ -1,88 +1,94 @@
from dash import Dash, Input, Output, State, ALL from __future__ import annotations
import dash_bootstrap_components as dbc
from dash_auth import BasicAuth
import logging import logging
from typing import List
from pydantic import ValidationError
import os 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 pydantic import ValidationError
from .layout import app_layout from .layout import app_layout
from src.database import ( from src.database import connect
connect, from src.database import get_visual_communication
get_visual_communication, from src.database import ModelOutputs
NoDocumentFoundException, from src.database import NoDocumentFoundException
upsert_annotations, from src.database import upsert_annotations
ModelOutputs
)
# setup app # setup app
app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP]) 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 app.layout = app_layout
server = app.server server = app.server
# setup authentication # setup authentication
AUTH_DICT = { 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) BasicAuth(app, AUTH_DICT)
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
# define callbacks # define callbacks
@app.callback( @app.callback(
Output("alert-element", "is_open"), Output('alert-element', 'is_open'),
Output("alert-element", "children"), Output('alert-element', 'children'),
Input("alert-message", "data") Input('alert-message', 'data'),
) )
def show_alert( def show_alert(
msg: str | None msg: str | None,
): ):
if msg is None or msg == "": if msg is None or msg == '':
return False, "" return False, ''
logging.info(f"updated alert message: {msg}") logging.info(f"updated alert message: {msg}")
return True, msg return True, msg
@app.callback( @app.callback(
Output("alert-message", "data"), Output('alert-message', 'data'),
Output("vis-com-name", "data"), Output('vis-com-name', 'data'),
Output("image-container", "src"), Output('image-container', 'src'),
Output({"type": "annotation", "index": ALL}, "value"), Output({'type': 'annotation', 'index': ALL}, 'value'),
Input("next-button", "n_clicks"), Input('next-button', 'n_clicks'),
State("vis-com-name", "data"), State('vis-com-name', 'data'),
State("image-container", "src"), State('image-container', 'src'),
State({"type": "annotation", "index": ALL}, "id"), State({'type': 'annotation', 'index': ALL}, 'id'),
State({"type": "annotation", "index": ALL}, "value"), State({'type': 'annotation', 'index': ALL}, 'value'),
prevent_initial_call=True, prevent_initial_call=True,
) )
def cycle_visual_communication_data( def cycle_visual_communication_data(
n_clicks: int, n_clicks: int,
vis_com_name: str, vis_com_name: str,
image_src: str, image_src: str,
annotation_keys: List, annotation_keys: list,
annotation_values: List, annotation_values: list,
): ):
logging.info("began cycling visual communication data") logging.info('began cycling visual communication data')
global collection global collection
# prepare default response # prepare default response
response = [ response = [
"", '',
vis_com_name, vis_com_name,
image_src, image_src,
annotation_values annotation_values,
] ]
# check if next-button clicked # check if next-button clicked
if n_clicks == 0: 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 return response
# check if visual communication name is set # check if visual communication name is set
if len(vis_com_name) > 0: 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: try:
# extract option keys # extract option keys
annotation_keys = [ annotation_keys = [
elem["index"] elem['index']
for elem in annotation_keys for elem in annotation_keys
] ]
# ensure all options are set # ensure all options are set
@@ -101,18 +107,18 @@ def cycle_visual_communication_data(
for elem for elem
in annotation_values in annotation_values
] ]
annotations = { annotation_map = {
key: value key: value
for key, value for key, value
in zip(annotation_keys, annotation_values) in zip(annotation_keys, annotation_values)
} }
# instantiate ModelOutputs object # instantiate ModelOutputs object
annotations = ModelOutputs.from_annotations(**annotations) annotations = ModelOutputs.from_annotations(**annotation_map)
# save data to # save data to database
upsert_annotations( upsert_annotations(
collection=collection, collection=collection,
vis_com_name=vis_com_name, vis_com_name=vis_com_name,
annotations=annotations annotations=annotations,
) )
except (ValueError, ValidationError) as exc: except (ValueError, ValidationError) as exc:
msg = f"failed saving annotation: {exc}" msg = f"failed saving annotation: {exc}"
@@ -120,12 +126,12 @@ def cycle_visual_communication_data(
response[0] = msg response[0] = msg
return tuple(response) return tuple(response)
# get new visual communication # get new visual communication
logging.info("trying to get new visual communication") logging.info('trying to get new visual communication')
try: try:
# get data # get data
vis_com = get_visual_communication( vis_com = get_visual_communication(
collection=collection, collection=collection,
with_annotation=False with_annotation=False,
) )
# set variables # set variables
vis_com_name = vis_com.name vis_com_name = vis_com.name
@@ -137,7 +143,7 @@ def cycle_visual_communication_data(
# reset annotations # reset annotations
annotation_values = [None for elem in annotation_values] annotation_values = [None for elem in annotation_values]
except NoDocumentFoundException: except NoDocumentFoundException:
msg = f"no unannotated data in database" msg = 'no unannotated data in database'
logging.warning(msg) logging.warning(msg)
response[0] = msg response[0] = msg
return tuple(response) return tuple(response)
@@ -145,5 +151,5 @@ def cycle_visual_communication_data(
response[1] = vis_com_name response[1] = vis_com_name
response[2] = image_src response[2] = image_src
response[3] = annotation_values 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) return tuple(response)
src/web/layout/__init__.py
+1 -1
View File
@@ -1 +1 @@
from .layout import app_layout from .layout import app_layout
src/web/layout/alerts.py
+1 -1
View File
@@ -1,4 +1,4 @@
from dash import html, dcc from dash import html
import dash_bootstrap_components as dbc import dash_bootstrap_components as dbc
src/web/layout/body.py
+1 -3
View File
@@ -1,6 +1,4 @@
import dash_mantine_components as dmc import dash_mantine_components as dmc
from dash import dcc, html
from typing import List
from .image import image_element from .image import image_element
from .inputs import inputs_element from .inputs import inputs_element
@@ -17,4 +15,4 @@ body_element = dmc.Container(
], ],
) )
], ],
) )
src/web/layout/inputs.py
+1 -1
View File
@@ -20,4 +20,4 @@ inputs_element = dmc.SimpleGrid(
labels_element, labels_element,
next_button next_button
] ]
) )
src/web/layout/labels.py
+58 -45
View File
@@ -1,26 +1,23 @@
from dash import html, dcc from __future__ import annotations
import dash_mantine_components as dmc import dash_mantine_components as dmc
from typing import List from dash import dcc
from dash import html
from src.model_experiential import ( from src.model_experiential import VisualSyntaxModelOutput
VisualSyntaxModelOutput from src.model_interpersonal import AngleModelOutput
) from src.model_interpersonal import ContactModelOutput
from src.model_interpersonal import ( from src.model_interpersonal import DistanceModelOutput
ContactModelOutput, from src.model_interpersonal import ModalityColorModelOutput
AngleModelOutput, from src.model_interpersonal import ModalityDepthModelOutput
PointOfViewModelOutput, from src.model_interpersonal import ModalityLightingModelOutput
DistanceModelOutput, from src.model_interpersonal import PointOfViewModelOutput
ModalityLightingModelOutput, from src.model_textual import FramingModelOutput
ModalityColorModelOutput, from src.model_textual import InformationValueModelOutput
ModalityDepthModelOutput from src.model_textual import SalienceModelOutput
)
from src.model_textual import (
InformationValueModelOutput,
FramingModelOutput,
SalienceModelOutput
)
def generate_option_labels(model) -> List[str]:
def generate_option_labels(model) -> list[str]:
"""Generate presentable list of attributes from an OutputModel.""" """Generate presentable list of attributes from an OutputModel."""
labels = [ labels = [
label.replace('_', ' ').title() label.replace('_', ' ').title()
@@ -28,44 +25,60 @@ def generate_option_labels(model) -> List[str]:
] ]
return labels return labels
def generate_visual_syntax_options_map(): def generate_visual_syntax_options_map():
"""Generate map of titles and options for visual syntax labels.""" """Generate map of titles and options for visual syntax labels."""
options_map = {} options_map = {}
# add experiential labels # add experiential labels
options_map["visual syntax"] = generate_option_labels(VisualSyntaxModelOutput) options_map['visual syntax'] = generate_option_labels(
VisualSyntaxModelOutput,
)
return options_map return options_map
def generate_interpersonal_options_map(): def generate_interpersonal_options_map():
"""Generate map of titles and options for interpersonal labels.""" """Generate map of titles and options for interpersonal labels."""
options_map = {} options_map = {}
# add interpersonal labels # add interpersonal labels
options_map["contact"] = generate_option_labels(ContactModelOutput) options_map['contact'] = generate_option_labels(ContactModelOutput)
options_map["angle"] = generate_option_labels(AngleModelOutput) options_map['angle'] = generate_option_labels(AngleModelOutput)
options_map["point of view"] = generate_option_labels(PointOfViewModelOutput) options_map['point of view'] = generate_option_labels(
options_map["distance"] = generate_option_labels(DistanceModelOutput) PointOfViewModelOutput,
options_map["modality lighting"] = generate_option_labels(ModalityLightingModelOutput) )
options_map["modality color"] = generate_option_labels(ModalityColorModelOutput) options_map['distance'] = generate_option_labels(DistanceModelOutput)
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 return options_map
def generate_textual_options_map(): def generate_textual_options_map():
"""Generate map of titles and options for textual labels.""" """Generate map of titles and options for textual labels."""
options_map = {} options_map = {}
# add textual labels # add textual labels
options_map["information value"] = generate_option_labels(InformationValueModelOutput) options_map['information value'] = generate_option_labels(
options_map["framing"] = generate_option_labels(FramingModelOutput) InformationValueModelOutput,
options_map["salience"] = generate_option_labels(SalienceModelOutput) )
options_map['framing'] = generate_option_labels(FramingModelOutput)
options_map['salience'] = generate_option_labels(SalienceModelOutput)
return options_map return options_map
# prepare experiential container # prepare experiential container
experiential_map = generate_visual_syntax_options_map() experiential_map = generate_visual_syntax_options_map()
experiential_container = dmc.Col( experiential_container = dmc.Col(
children=[ children=[
html.H4("experiential".title()), html.H4('experiential'.title()),
], span=5 ], span=5,
) )
for title, options in experiential_map.items(): 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( experiential_container.children.append(
dmc.Container([ dmc.Container([
html.B(title.title()), html.B(title.title()),
@@ -73,17 +86,17 @@ for title, options in experiential_map.items():
options=options, options=options,
id=id_dict, id=id_dict,
), ),
]) ]),
) )
# prepare interpersonal container # prepare interpersonal container
interpersonal_map = generate_interpersonal_options_map() interpersonal_map = generate_interpersonal_options_map()
interpersonal_container = dmc.Col( interpersonal_container = dmc.Col(
children=[ children=[
html.H4("interpersonal".title()), html.H4('interpersonal'.title()),
], span=3 ], span=3,
) )
for title, options in interpersonal_map.items(): 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( interpersonal_container.children.append(
dmc.Container([ dmc.Container([
html.B(title.title()), html.B(title.title()),
@@ -91,17 +104,17 @@ for title, options in interpersonal_map.items():
options=options, options=options,
id=id_dict, id=id_dict,
), ),
]) ]),
) )
# prepare textual container # prepare textual container
textual_map = generate_textual_options_map() textual_map = generate_textual_options_map()
textual_container = dmc.Col( textual_container = dmc.Col(
children=[ children=[
html.H4("textual".title()), html.H4('textual'.title()),
], span=4 ], span=4,
) )
for title, options in textual_map.items(): 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( textual_container.children.append(
dmc.Container([ dmc.Container([
html.B(title), html.B(title),
@@ -109,7 +122,7 @@ for title, options in textual_map.items():
options=options, options=options,
id=id_dict, id=id_dict,
), ),
]) ]),
) )
labels_element = dmc.Grid( labels_element = dmc.Grid(
@@ -117,5 +130,5 @@ labels_element = dmc.Grid(
experiential_container, experiential_container,
interpersonal_container, interpersonal_container,
textual_container, textual_container,
] ],
) )
src/web/layout/layout.py
-1
View File
@@ -1,4 +1,3 @@
from dash import dcc
import dash_mantine_components as dmc import dash_mantine_components as dmc
from .stores import stores_element from .stores import stores_element
src/web/layout/stores.py
+5 -3
View File
@@ -5,12 +5,14 @@ import os
storage_type = "session" storage_type = "session"
if "ENV" in os.environ and os.getenv("ENV") == "DEV": if "ENV" in os.environ and os.getenv("ENV") == "DEV":
storage_type = "memory" 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( stores_element = html.Div(
children=[ children=[
dcc.Store(id="alert-message", storage_type=storage_type, data=""), dcc.Store(id="alert-message", storage_type=storage_type, data=""),
dcc.Store(id="vis-com-name", storage_type=storage_type, data=""), dcc.Store(id="vis-com-name", storage_type=storage_type, data=""),
] ]
) )
tests/generate_random_prediction_test.py
+24
View File
@@ -0,0 +1,24 @@
from __future__ import annotations
from pathlib import Path
from database import VisualCommunication
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()]
print(img_path_list)
# instantiate data object
vis_com_list = [
VisualCommunication.from_file(path)
for path
in img_path_list
]
# generate random predictions
for vis_com in vis_com_list:
vis_com.generate_random_prediction()
for vis_com in vis_com_list:
print(vis_com)
tests/test_get_visual_communication.py → tests/get_visual_communication_test.py
+9 -9
View File
@@ -1,18 +1,18 @@
from pathlib import Path from __future__ import annotations
from dotenv import load_dotenv
import os import os
from pathlib import Path
from src.database import ( from database import connect
connect, from database import get_visual_communication
get_visual_communication from dotenv import load_dotenv
)
if __name__ == "__main__": if __name__ == '__main__':
# prepare env vars # prepare env vars
env_path = Path(__file__).parent.parent / "local.env" env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists() assert env_path.exists()
load_dotenv(env_path) load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
print(client.server_info()) print(client.server_info())
tests/test_image_download.py → tests/image_download_test.py
+10 -7
View File
@@ -1,15 +1,18 @@
from pathlib import Path from __future__ import annotations
from dotenv import load_dotenv
import os 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 # prepare env vars
env_path = Path(__file__).parent.parent / "local.env" env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists() assert env_path.exists()
load_dotenv(env_path) load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
print(client.server_info()) print(client.server_info())
@@ -17,7 +20,7 @@ if __name__ == "__main__":
data = None data = None
for data in collection.find().limit(3): for data in collection.find().limit(3):
if data is None: if data is None:
print("no document found") print('no document found')
break break
vis_com: VisualCommunication = VisualCommunication.model_validate(data) vis_com: VisualCommunication = VisualCommunication.model_validate(data)
print(repr(vis_com)) print(repr(vis_com))
tests/test_image_upload.py → tests/image_upload_test.py
+17 -10
View File
@@ -1,25 +1,32 @@
from __future__ import annotations
import os
from pathlib import Path from pathlib import Path
from database import connect
from database import VisualCommunication
from dotenv import load_dotenv from dotenv import load_dotenv
from pymongo.errors import DuplicateKeyError from pymongo.errors import DuplicateKeyError
import os
from src.database import VisualCommunication, connect if __name__ == '__main__':
if __name__ == "__main__":
# get list of image paths # get list of image paths
test_dir = Path(__file__).parent test_dir = Path(__file__).parent
img_dir = test_dir / "imgs" img_dir = test_dir / 'imgs'
img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()] img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
print(img_path_list) print(img_path_list)
# instantiate data object # instantiate data object
vis_com_list = [VisualCommunication.from_file(path) for path in img_path_list] vis_com_list = [
VisualCommunication.from_file(path)
for path
in img_path_list
]
for vis_com in vis_com_list: for vis_com in vis_com_list:
print(repr(vis_com)) print(repr(vis_com))
# prepare env vars # prepare env vars
env_path = test_dir.parent / "local.env" env_path = test_dir.parent / 'local.env'
assert env_path.exists() assert env_path.exists()
load_dotenv(env_path) load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
print(client.server_info()) print(client.server_info())
@@ -28,6 +35,6 @@ if __name__ == "__main__":
try: try:
result = collection.insert_one(vis_com.model_dump()) result = collection.insert_one(vis_com.model_dump())
except DuplicateKeyError as exc: except DuplicateKeyError as exc:
print("ignoring:\n", exc) print('ignoring:\n', exc)
else: else:
print(f"inserted document: {result}") print(f"inserted document: {result}")
tests/model_outputs_from_annotation_test.py
+16
View File
@@ -0,0 +1,16 @@
from __future__ import annotations
from database import ModelOutputs
if __name__ == '__main__':
# instantiate data object
vis_com_list = [
ModelOutputs.from_random()
for i
in range(3)
]
# generate random predictions
[vis_com.generate_random_prediction() for vis_com in vis_com_list]
for vis_com in vis_com_list:
print(vis_com)
tests/test_prediction_upload.py → tests/prediction_upload_test.py
+22 -15
View File
@@ -1,15 +1,15 @@
from pathlib import Path from __future__ import annotations
from dotenv import load_dotenv
import os
import logging import logging
import os
from pathlib import Path
from src.database import ( from database import connect
VisualCommunication, from database import upsert_predictions
connect, from database import VisualCommunication
upsert_predictions from dotenv import load_dotenv
)
if __name__ == "__main__": if __name__ == '__main__':
# setup logging # setup logging
fmt = ( fmt = (
'%(asctime)s | ' '%(asctime)s | '
@@ -22,21 +22,28 @@ if __name__ == "__main__":
logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO) logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
# get list of image paths # get list of image paths
test_dir = Path(__file__).parent test_dir = Path(__file__).parent
img_dir = test_dir / "imgs" img_dir = test_dir / 'imgs'
img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()] img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
# instantiate data object # instantiate data object
vis_com_list = [VisualCommunication.from_file(path) for path in img_path_list] vis_com_list = [
VisualCommunication.from_file(path)
for path
in img_path_list
]
# generate random predictions # 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 # prepare env vars
env_path = test_dir.parent / "local.env" env_path = test_dir.parent / 'local.env'
assert env_path.exists() assert env_path.exists()
load_dotenv(env_path) load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
# upload visual communication # upload visual communication
for vis_com in vis_com_list: for vis_com in vis_com_list:
if vis_com.prediction is None:
continue
upsert_predictions( upsert_predictions(
collection=collection, collection=collection,
vis_com_name=vis_com.name, vis_com_name=vis_com.name,
tests/test_generate_random_prediction.py
-17
View File
@@ -1,17 +0,0 @@
from pathlib import Path
from src.database import VisualCommunication
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()]
print(img_path_list)
# instantiate data object
vis_com_list = [VisualCommunication.from_file(path) for path 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:
print(vis_com)
tests/test_model_outputs_from_annotation.py
-13
View File
@@ -1,13 +0,0 @@
from src.database import ModelOutputs
if __name__ == "__main__":
# instantiate data object
annotation = {
}
vis_com_list = [ModelOutputs.from_annotation(path) for path 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:
print(vis_com)
tests/test_total_documents.py
-20
View File
@@ -1,20 +0,0 @@
from pathlib import Path
from dotenv import load_dotenv
import os
from src.database import (
connect,
total_documents
)
if __name__ == "__main__":
# prepare env vars
env_path = Path(__file__).parent.parent / "local.env"
assert env_path.exists()
load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost"
# connect to database
collection, db, client = connect()
# get visual communication
num_docs = total_documents(collection)
print(f"total number of documents in database: {num_docs}")
tests/test_total_annotated.py → tests/total_annotated_test.py
+10 -10
View File
@@ -1,20 +1,20 @@
from pathlib import Path from __future__ import annotations
from dotenv import load_dotenv
import os import os
from pathlib import Path
from src.database import ( from database import connect
connect, from database import total_annotated
total_annotated from dotenv import load_dotenv
)
if __name__ == "__main__": if __name__ == '__main__':
# prepare env vars # prepare env vars
env_path = Path(__file__).parent.parent / "local.env" env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists() assert env_path.exists()
load_dotenv(env_path) load_dotenv(env_path)
os.environ["MONGO_HOST"] = "localhost" os.environ['MONGO_HOST'] = 'localhost'
# connect to database # connect to database
collection, db, client = connect() collection, db, client = connect()
# get visual communication # get visual communication
num_docs = total_annotated(collection) num_docs = total_annotated(collection)
print(f"number of annotated documents in database: {num_docs}") print(f"number of annotated documents in database: {num_docs}")
tests/total_documents_test.py
+20
View File
@@ -0,0 +1,20 @@
from __future__ import annotations
import os
from pathlib import Path
from database import connect
from database import total_documents
from dotenv import load_dotenv
if __name__ == '__main__':
# prepare env vars
env_path = Path(__file__).parent.parent / 'local.env'
assert env_path.exists()
load_dotenv(env_path)
os.environ['MONGO_HOST'] = 'localhost'
# connect to database
collection, db, client = connect()
# get visual communication
num_docs = total_documents(collection)
print(f"total number of documents in database: {num_docs}")