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

add_gitea_workflows #17

Merged
brian merged 48 commits from add_gitea_workflows into main 2024-02-25 18:45:08 +01:00
Conversation 0 Commits 48 Files Changed 37
+667 -330
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)"]
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]]
name = "flask"
version = "3.0.2"
@@ -456,6 +472,74 @@ files = [
{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]]
name = "nest-asyncio"
version = "1.6.0"
@@ -578,6 +662,17 @@ files = [
packaging = "*"
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]]
name = "pydantic"
version = "2.6.1"
@@ -688,6 +783,17 @@ files = [
[package.dependencies]
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]]
name = "pymongo"
version = "4.6.1"
@@ -881,6 +987,17 @@ files = [
[package.extras]
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]]
name = "typing-extensions"
version = "4.9.0"
@@ -944,4 +1061,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
[metadata]
lock-version = "2.0"
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"
[tool.poetry.group.test.dependencies]
flake8 = "^7.0.0"
mypy = "^1.8.0"
types-pillow = "^10.2.0.20240213"
[build-system]
requires = ["poetry-core"]
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 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 PIL import Image
from pydantic import BaseModel
from pydantic import field_serializer
from pydantic import field_validator
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
VisualSyntaxModelOutput,
)
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):
visual_syntax: VisualSyntaxModelOutput
@@ -39,7 +43,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())
@@ -47,12 +51,12 @@ class ModelOutputs(BaseModel):
def from_random(cls) -> ModelOutputs:
"""Instantiate with random numbers."""
kwargs = {
field: field_info.annotation.from_random()
field: field_info.annotation.from_random() # type: ignore
for field, field_info
in cls.model_fields.items()
}
return cls(**kwargs)
@classmethod
def from_annotations(
cls,
@@ -66,21 +70,32 @@ class ModelOutputs(BaseModel):
modality_depth: str,
information_value: str,
framing: str,
salience: str
salience: str,
) -> 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)
@@ -107,15 +122,18 @@ class VisualCommunication(BaseModel):
image.load()
return VisualCommunication(name=name, image=image)
@field_serializer("image")
def serialize_image(image: Image.Image) -> bytes:
@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:
def convert_to_image(
cls,
image: Image.Image | BytesIO | bytes,
) -> Image.Image:
if isinstance(image, bytes):
image = BytesIO(image)
if isinstance(image, BytesIO):
@@ -124,20 +142,17 @@ 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
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()
class NoDocumentFoundException(Exception):
pass
src/database/utils.py
+24 -13
View File
@@ -20,32 +20,39 @@ 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:
query["annotation"] = {"$ne": None}
else:
query["annotation"] = None
query["annotation"] = {"$eq": 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:
data_list = list(data) # read data from cursor object
if len(data_list) == 0:
logging.error("failed getting visual communication")
raise NoDocumentFoundException()
data = data[0]
logging.info("finished")
return VisualCommunication.model_validate(data)
return VisualCommunication.model_validate(data_list[0])
def upsert_predictions(
@@ -58,7 +65,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 +88,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,
src/main.py
+19 -17
View File
@@ -1,22 +1,26 @@
import logging
from dotenv import load_dotenv
from pathlib import Path
from __future__ import annotations
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
from src.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
@@ -37,14 +41,12 @@ fmt = (
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
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')
src/model_experiential/__init__.py
-23
View File
@@ -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"
# ]
src/model_experiential/classes.py
+12 -6
View File
@@ -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()
src/model_interpersonal/__init__.py
+1 -1
View File
@@ -6,4 +6,4 @@ from .classes import (
ModalityLightingModelOutput,
ModalityColorModelOutput,
ModalityDepthModelOutput
)
)
src/model_interpersonal/classes.py
+11 -6
View File
@@ -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()
src/model_textual/__init__.py
+1 -1
View File
@@ -2,4 +2,4 @@ from .classes import (
InformationValueModelOutput,
FramingModelOutput,
SalienceModelOutput
)
)
src/model_textual/classes.py
+12 -7
View File
@@ -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())
src/web/__init__.py
+4 -4
View File
@@ -1,4 +1,4 @@
from .app import (
app,
server
)
from __future__ import annotations
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
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 pydantic import ValidationError
from .layout import app_layout
from src.database import (
connect,
get_visual_communication,
NoDocumentFoundException,
upsert_annotations,
ModelOutputs
)
from src.database import connect
from src.database import get_visual_communication
from src.database import ModelOutputs
from src.database import NoDocumentFoundException
from src.database import upsert_annotations
# 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)
# connect to database
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")
if n_clicks == 0:
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
@@ -101,18 +107,18 @@ def cycle_visual_communication_data(
for elem
in annotation_values
]
annotations = {
key: value
for key, value
annotation_map = {
key: value
for key, value
in zip(annotation_keys, annotation_values)
}
# instantiate ModelOutputs object
annotations = ModelOutputs.from_annotations(**annotations)
# save data to
annotations = ModelOutputs.from_annotations(**annotation_map)
# save data to database
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}"
@@ -120,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
@@ -137,7 +143,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)
@@ -145,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)
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
src/web/layout/body.py
+1 -3
View File
@@ -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(
],
)
],
)
)
src/web/layout/inputs.py
+1 -1
View File
@@ -20,4 +20,4 @@ inputs_element = dmc.SimpleGrid(
labels_element,
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
from typing import List
from dash import dcc
from dash import html
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 src.model_experiential import VisualSyntaxModelOutput
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
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()
@@ -28,44 +25,60 @@ 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["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['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['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["framing"] = generate_option_labels(FramingModelOutput)
options_map["salience"] = generate_option_labels(SalienceModelOutput)
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(
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()),
@@ -73,17 +86,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()),
@@ -91,17 +104,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),
@@ -109,7 +122,7 @@ for title, options in textual_map.items():
options=options,
id=id_dict,
),
])
]),
)
labels_element = dmc.Grid(
@@ -117,5 +130,5 @@ labels_element = dmc.Grid(
experiential_container,
interpersonal_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
from .stores import stores_element
src/web/layout/stores.py
+5 -3
View File
@@ -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=""),
]
)
)
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 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())
tests/test_image_download.py → tests/image_download_test.py
+10 -7
View File
@@ -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))
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 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 = [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:
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())
@@ -28,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}")
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 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,21 +22,28 @@ 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) for path in img_path_list]
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:
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,
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 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
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}")