Coverage for src\cognitivefactory\interactive_clustering

1# -*- coding: utf-8 -*-

3"""

4* Name: cognitivefactory.interactive_clustering_gui.app

5* Description: Definition of FastAPI application and routes for interactive clustering graphical user interface.

6* Author: Erwan Schild

7* Created: 22/10/2021

8* Licence: CeCILL-C License v1.0 (https://cecill.info/licences.fr.html)

9"""

11# ==============================================================================

12# IMPORT PYTHON DEPENDENCIES

13# ==============================================================================

15import html

16import json

17import os

18import pathlib

19import shutil

20from datetime import datetime

21from typing import Any, Callable, Dict, List, Optional, Tuple

23import pandas as pd

24from dateutil import tz

25from fastapi import (

26 BackgroundTasks,

27 Body,

28 FastAPI,

29 File,

30 HTTPException,

31 Path,

32 Query,

33 Request,

34 Response,

35 UploadFile,

36 status,

37)

38from fastapi.middleware.cors import CORSMiddleware

39from fastapi.responses import FileResponse # HTMLResponse

40from fastapi.staticfiles import StaticFiles

41from fastapi.templating import Jinja2Templates

42from filelock import FileLock

43from importlib_metadata import version

44from prometheus_client import Gauge

45from prometheus_fastapi_instrumentator import Instrumentator, metrics

46from zipp import zipfile

48from cognitivefactory.interactive_clustering_gui import backgroundtasks

49from cognitivefactory.interactive_clustering_gui.models.queries import (

50 ConstraintsSortOptions,

51 ConstraintsValues,

52 TextsSortOptions,

53)

54from cognitivefactory.interactive_clustering_gui.models.settings import (

55 ClusteringSettingsModel,

56 ICGUISettings,

57 PreprocessingSettingsModel,

58 SamplingSettingsModel,

59 VectorizationSettingsModel,

60 default_ClusteringSettingsModel,

61 default_PreprocessingSettingsModel,

62 default_SamplingSettingsModel,

63 default_VectorizationSettingsModel,

64)

65from cognitivefactory.interactive_clustering_gui.models.states import ICGUIStates, get_ICGUIStates_details

67# ==============================================================================

68# CONFIGURE FASTAPI APPLICATION

69# ==============================================================================

71# Create an FastAPI instance of Interactive Clustering Graphical User Interface.

72app = FastAPI(

73 title="Interactive Clustering GUI",

74 description="Web application for Interactive Clustering methodology",

75 version=version("cognitivefactory-interactive-clustering-gui"),

76)

78# Add Cross-Origin Resource Sharing

79app.add_middleware(

80 CORSMiddleware,

81 allow_origins=[ # origins for which we allow CORS (Cross-Origin Resource Sharing).

82 "http://localhost:8000", # locally served docs.

83 # "https://cognitivefactory.github.io/interactive-clustering-gui/", # deployed docs.

84 # you could add `http://your_vm_hostname:8000`.

85 ],

86)

88# Mount `css`, `javascript` and `resources` files.

89app.mount("/css", StaticFiles(directory=pathlib.Path(__file__).parent / "css"), name="css")

90app.mount("/js", StaticFiles(directory=pathlib.Path(__file__).parent / "js"), name="js")

91app.mount("/resources", StaticFiles(directory=pathlib.Path(__file__).parent / "resources"), name="resources")

93# Define `DATA_DIRECTORY` (where data are stored).

94DATA_DIRECTORY = pathlib.Path(os.environ.get("DATA_DIRECTORY", ".data"))

95DATA_DIRECTORY.mkdir(parents=True, exist_ok=True)

98# ==============================================================================

99# CONFIGURE JINJA2 TEMPLATES

100# ==============================================================================

101

102# Define HTML templates to render.

103templates = Jinja2Templates(directory=pathlib.Path(__file__).parent / "html")

104

105

106# Define function to convert timestamp to date.

107def timestamp_to_date(timestamp: float, timezone_str: str = "Europe/Paris") -> str:

108 """

109 From timestamp to date.

110

111 Args:

112 timestamp (float): The timstamp to convert.

113 timezone_str (str, optional): The time zone. Defaults to `"Europe/Paris"`.

114

115 Returns:

116 str: The requested date.

117 """

118 timezone = tz.gettz(timezone_str)

119 return datetime.fromtimestamp(timestamp, timezone).strftime("%d/%m/%Y")

120

121

122templates.env.filters["timestamp_to_date"] = timestamp_to_date

123

124

125# Define function to convert timestamp to hour.

126def timestamp_to_hour(timestamp: float, timezone_str: str = "Europe/Paris") -> str:

127 """

128 From timestamp to hours.

129

130 Args:

131 timestamp (float): The timstamp to convert.

132 timezone_str (str, optional): The time zone. Defaults to `"Europe/Paris"`.

133

134 Returns:

135 str: The requested hour.

136 """

137 timezone = tz.gettz(timezone_str)

138 return datetime.fromtimestamp(timestamp, timezone).strftime("%H:%M:%S")

139

140

141templates.env.filters["timestamp_to_hour"] = timestamp_to_hour

142

143

144# Define function to get previous key in a dictionary.

145def get_previous_key(key: str, dictionary: Dict[str, Any]) -> Optional[str]:

146 """

147 Get previous key in a dictionary.

148

149 Args:

150 key (str): The current key.

151 dictionary (Dict[str, Any]): The dictionary.

152

153 Returns:

154 Optional[str]: The previous key.

155 """

156 list_of_keys: List[str] = list(dictionary.keys())

157 if key in list_of_keys:

158 previous_key_index: int = list_of_keys.index(key) - 1

159 return list_of_keys[previous_key_index] if 0 <= previous_key_index else None

160 return None

161

162

163templates.env.filters["get_previous_key"] = get_previous_key

164

165

166# Define function to get next key in a dictionary.

167def get_next_key(key: str, dictionary: Dict[str, Any]) -> Optional[str]:

168 """

169 Get next key in a dictionary.

170

171 Args:

172 key (str): The current key.

173 dictionary (Dict[str, Any]): The dictionary.

174

175 Returns:

176 Optional[str]: The next key.

177 """

178 list_of_keys: List[str] = list(dictionary.keys())

179 if key in list_of_keys:

180 next_key_index: int = list_of_keys.index(key) + 1

181 return list_of_keys[next_key_index] if next_key_index < len(list_of_keys) else None

182 return None

183

184

185templates.env.filters["get_next_key"] = get_next_key

186

187

188# ==============================================================================

189# CONFIGURE FASTAPI METRICS

190# ==============================================================================

191

192

193def prometheus_disk_usage() -> Callable[[metrics.Info], None]:

194 """

195 Define a metric of disk usage.

196

197 Returns:

198 Callable[[metrics.Info], None]: instrumentation.

199 """

200 gaugemetric = Gauge(

201 "disk_usage",

202 "The disk usage in %",

203 )

204

205 def instrumentation(info: metrics.Info) -> None: # noqa: WPS430 (nested function)

206 total, used, _ = shutil.disk_usage(DATA_DIRECTORY)

207 gaugemetric.set(used * 100 / total)

208

209 return instrumentation

210

211

212# Define application instrumentator and add metrics.

213instrumentator = Instrumentator()

214instrumentator.add(metrics.default())

215instrumentator.add(prometheus_disk_usage())

216instrumentator.instrument(app)

217instrumentator.expose(app)

218

219

220# ==============================================================================

221# DEFINE STATE ROUTES FOR APPLICATION STATE

222# ==============================================================================

223

224

225###

226### STATE: Startup event.

227###

228@app.on_event("startup")

229async def startup() -> None: # pragma: no cover

230 """Startup event."""

231

232 # Initialize ready state.

233 app.state.ready = False

234

235 # Apply database connection, long loading, etc.

236

237 # Update ready state when done.

238 app.state.ready = True

239

240

241###

242### STATE: Check if app is ready.

243###

244@app.get(

245 "/ready",

246 tags=["app state"],

247 status_code=status.HTTP_200_OK,

248)

249async def ready() -> Response: # pragma: no cover

250 """

251 Tell if the API is ready.

252

253 Returns:

254 An HTTP response with either 200 or 503 codes.

255 """

256

257 # Return 200_OK if ready.

258 if app.state.ready:

259 return Response(status_code=status.HTTP_200_OK)

260

261 # Return 503_SERVICE_UNAVAILABLE otherwise.

262 return Response(status_code=status.HTTP_503_SERVICE_UNAVAILABLE)

263

264

265###

266### STATE: Check if app is alive.

267###

268@app.get(

269 "/alive",

270 tags=["app state"],

271 status_code=status.HTTP_200_OK,

272)

273async def alive() -> Response: # pragma: no cover

274 """

275 Tell if the API is alive/healthy.

276

277 Returns:

278 Response: An HTTP response with either 200 or 503 codes.

279 """

280

281 try:

282 # Assert the volume can be reached.

283 pathlib.Path(DATA_DIRECTORY / ".available").touch()

284 # Or anything else asserting the API is healthy.

285 except OSError:

286 return Response(status_code=status.HTTP_503_SERVICE_UNAVAILABLE)

287 return Response(status_code=status.HTTP_200_OK)

288

289

290# ==============================================================================

291# DEFINE ROUTES FOR HOME AND DOCUMENTATION

292# ==============================================================================

293

294

295###

296### ROUTE: Get HTML welcome page with projects listings.

297###

298@app.get(

299 "/",

300 tags=["Home and Documentation"],

301 response_class=Response,

302 status_code=status.HTTP_200_OK,

303)

304async def get_html_welcome_page(

305 request: Request,

306) -> Response:

307 """

308 Define HTML welcome page with projects listings.

309

310 Args:

311 request (Request): The request context.

312

313 Returns:

314 Response: The requested page.

315 """

316

317 # Return HTML welcome page.

318 return templates.TemplateResponse(

319 name="welcome.html",

320 context={

321 "request": request,

322 # Get projects and their description.

323 "projects": {

324 project_id: {

325 "metadata": (await get_metadata(project_id=project_id))["metadata"],

326 "status": (await get_status(project_id=project_id))["status"],

327 }

328 for project_id in (await get_projects())

329 },

330 },

331 status_code=status.HTTP_200_OK,

332 )

333

334

335###

336### ROUTE: Get HTML help page.

337###

338@app.get(

339 "/gui/help",

340 tags=["Home and Documentation"],

341 response_class=Response,

342 status_code=status.HTTP_200_OK,

343)

344async def get_html_help_page(

345 request: Request,

346) -> Response:

347 """

348 Get HTML help page.

349

350 Args:

351 request (Request): The request context.

352

353 Returns:

354 Response: The requested page.

355 """

356

357 # Return HTML help page.

358 return templates.TemplateResponse(

359 name="help.html",

360 context={

361 "request": request,

362 },

363 status_code=status.HTTP_200_OK,

364 )

365

366

367# ==============================================================================

368# DEFINE ROUTES FOR PROJECT MANAGEMENT

369# ==============================================================================

370

371

372###

373### ROUTE: Get the list of existing project IDs.

374###

375@app.get(

376 "/api/projects",

377 tags=["Projects"],

378 status_code=status.HTTP_200_OK,

379)

380async def get_projects() -> List[str]:

381 """

382 Get the list of existing project IDs.

383 (A project is represented by a subfolder in `.data` folder.)

384

385 Returns:

386 List[str]: The list of existing project IDs.

387 """

388

389 # Return the list of project IDs.

390 return [project_id for project_id in os.listdir(DATA_DIRECTORY) if os.path.isdir(DATA_DIRECTORY / project_id)]

391

392

393###

394### ROUTE: Create a project.

395###

396@app.post(

397 "/api/projects",

398 tags=["Projects"],

399 status_code=status.HTTP_201_CREATED,

400)

401async def create_project(

402 project_name: str = Query(

403 ...,

404 description="The name of the project. Should not be an empty string.",

405 min_length=3,

406 max_length=64,

407 ),

408 dataset_file: UploadFile = File(

409 ...,

410 description="The dataset file to load. Use a `.csv` (`;` separator) or `.xlsx` file. Please use a list of texts in the first column, without header, with encoding 'utf-8'.",

411 # TODO: max_size="8MB",

412 ),

413) -> Dict[str, Any]:

414 """

415 Create a project.

416

417 Args:

418 project_name (str): The name of the project. Should not be an empty string.

419 dataset_file (UploadFile): The dataset file to load. Use a `.csv` (`;` separator) or `.xlsx` file. Please use a list of texts in the first column, without header, with encoding 'utf-8'.

420

421 Raises:

422 HTTPException: Raises `HTTP_400_BAD_REQUEST` if parameters `project_name` or `dataset_file` are invalid.

423

424 Returns:

425 Dict[str, Any]: A dictionary that contains the ID of the created project.

426 """

427

428 # Define the new project ID.

429 current_timestamp: float = datetime.now().timestamp()

430 current_project_id: str = str(int(current_timestamp * 10**6))

431

432 # Check project name.

433 if project_name.strip() == "":

434 raise HTTPException(

435 status_code=status.HTTP_400_BAD_REQUEST,

436 detail="The project name '{project_name_str}' is invalid.".format(

437 project_name_str=str(project_name),

438 ),

439 )

440

441 # Initialize variable to store loaded dataset.

442 list_of_texts: List[str] = []

443

444 # Load dataset: Case of `.csv` with `;` separator.

445 if dataset_file.content_type in {"text/csv", "application/vnd.ms-excel"}:

446 # "text/csv" == ".csv"

447 # "application/vnd.ms-excel" == ".xls"

448 try: # noqa: WPS229 # Found too long `try` body length

449 dataset_csv: pd.Dataframe = pd.read_csv(

450 filepath_or_buffer=dataset_file.file,

451 sep=";",

452 header=None, # No header expected in the csv file.

453 )

454 list_of_texts = dataset_csv[dataset_csv.columns[0]].tolist()

455 except Exception:

456 raise HTTPException(

457 status_code=status.HTTP_400_BAD_REQUEST,

458 detail="The dataset file is invalid. `.csv` file, with `;` separator, must contain a list of texts in the first column, with no header.",

459 )

460 # Load dataset: Case of `.xlsx`.

461 elif dataset_file.content_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":

462 # "application/vnd.ms-excel" == ".xlsx"

463 try: # noqa: WPS229 # Found too long `try` body length

464 dataset_xlsx: pd.Dataframe = pd.read_excel(

465 io=dataset_file.file.read(),

466 engine="openpyxl",

467 header=None, # No header expected in the xlsx file.

468 )

469 list_of_texts = dataset_xlsx[dataset_xlsx.columns[0]].tolist()

470 except Exception:

471 raise HTTPException(

472 status_code=status.HTTP_400_BAD_REQUEST,

473 detail="The dataset file is invalid. `.xlsx` file must contain a list of texts in the first column, with no header.",

474 )

475

476 # Load dataset: case of not supported type.

477 else:

478 raise HTTPException(

479 status_code=status.HTTP_400_BAD_REQUEST,

480 detail="The file type '{dataset_file_type}' is not supported. Please use '.csv' (`;` separator) or '.xlsx' file.".format(

481 dataset_file_type=str(dataset_file.content_type),

482 ),

483 )

484

485 # Create the directory and subdirectories of the new project.

486 os.mkdir(DATA_DIRECTORY / current_project_id)

487

488 # Initialize storage of metadata.

489 with open(DATA_DIRECTORY / current_project_id / "metadata.json", "w") as metadata_fileobject:

490 json.dump(

491 {

492 "project_id": current_project_id,

493 "project_name": str(project_name.strip()),

494 "creation_timestamp": current_timestamp,

495 },

496 metadata_fileobject,

497 indent=4,

498 )

499

500 # Initialize storage of status.

501 with open(DATA_DIRECTORY / current_project_id / "status.json", "w") as status_fileobject:

502 json.dump(

503 {

504 "iteration_id": 0, # Use string format for JSON serialization in dictionaries.

505 "state": ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION,

506 "task": None, # "progression", "detail".

507 },

508 status_fileobject,

509 indent=4,

510 )

511

512 # Initialize storage of texts.

513 with open(DATA_DIRECTORY / current_project_id / "texts.json", "w") as texts_fileobject:

514 json.dump(

515 {

516 str(i): {

517 "text_original": str(text), # Will never be changed.

518 "text": str(text), # Can be change by renaming.

519 "text_preprocessed": str(text), # Will be preprocessed during `Modelizationpdate` task.

520 "is_deleted": False,

521 }

522 for i, text in enumerate(list_of_texts)

523 },

524 texts_fileobject,

525 indent=4,

526 )

527

528 # Initialize storage of constraints.

529 with open(DATA_DIRECTORY / current_project_id / "constraints.json", "w") as constraints_fileobject:

530 json.dump(

531 {}, # Dict[str, Any]

532 constraints_fileobject,

533 indent=4,

534 )

535

536 # Initialize storage of modelization inference assignations.

537 with open(DATA_DIRECTORY / current_project_id / "modelization.json", "w") as modelization_fileobject:

538 json.dump(

539 {str(i): {"MUST_LINK": [str(i)], "CANNOT_LINK": [], "COMPONENT": i} for i in range(len(list_of_texts))},

540 modelization_fileobject,

541 indent=4,

542 )

543

544 # Initialize settings storage.

545 with open(DATA_DIRECTORY / current_project_id / "settings.json", "w") as settings_fileobject:

546 json.dump(

547 {

548 "0": {

549 "preprocessing": default_PreprocessingSettingsModel().to_dict(),

550 "vectorization": default_VectorizationSettingsModel().to_dict(),

551 "clustering": default_ClusteringSettingsModel().to_dict(),

552 },

553 },

554 settings_fileobject,

555 indent=4,

556 )

557

558 # Initialize storage of sampling results.

559 with open(DATA_DIRECTORY / current_project_id / "sampling.json", "w") as sampling_fileobject:

560 json.dump({}, sampling_fileobject, indent=4) # Dict[str, List[str]]

561

562 # Initialize storage of clustering results.

563 with open(DATA_DIRECTORY / current_project_id / "clustering.json", "w") as clustering_fileobject:

564 json.dump({}, clustering_fileobject, indent=4) # Dict[str, Dict[str, str]]

565

566 # Return the ID of the created project.

567 return {

568 "project_id": current_project_id,

569 "detail": (

570 "The project with name '{project_name_str}' has been created. It has the id '{project_id_str}'.".format(

571 project_name_str=str(project_name),

572 project_id_str=str(current_project_id),

573 )

574 ),

575 }

576

577

578###

579### ROUTE: Delete a project.

580###

581@app.delete(

582 "/api/projects/{project_id}",

583 tags=["Projects"],

584 status_code=status.HTTP_202_ACCEPTED,

585)

586async def delete_project(

587 project_id: str = Path(

588 ...,

589 description="The ID of the project to delete.",

590 ),

591) -> Dict[str, Any]:

592 """

593 Delete a project.

594

595 Args:

596 project_id (str): The ID of the project to delete.

597

598 Returns:

599 Dict[str, Any]: A dictionary that contains the ID of the deleted project.

600 """

601

602 # Delete the project.

603 if os.path.isdir(DATA_DIRECTORY / project_id):

604 shutil.rmtree(DATA_DIRECTORY / project_id, ignore_errors=True)

605

606 # Return the deleted project id.

607 return {

608 "project_id": project_id,

609 "detail": "The deletion of project with id '{project_id_str}' is accepted.".format(

610 project_id_str=str(project_id),

611 ),

612 }

613

614

615###

616### ROUTE: Get metadata.

617###

618@app.get(

619 "/api/projects/{project_id}/metadata",

620 tags=["Projects"],

621 status_code=status.HTTP_200_OK,

622)

623async def get_metadata(

624 project_id: str = Path(

625 ...,

626 description="The ID of the project.",

627 ),

628) -> Dict[str, Any]:

629 """

630 Get metadata.

631

632 Args:

633 project_id (str): The ID of the project.

634

635 Raises:

636 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

637

638 Returns:

639 Dict[str, Any]: A dictionary that contains metadata.

640 """

641

642 # Check project id.

643 if project_id not in (await get_projects()):

644 raise HTTPException(

645 status_code=status.HTTP_404_NOT_FOUND,

646 detail="The project with id '{project_id_str}' doesn't exist.".format(

647 project_id_str=str(project_id),

648 ),

649 )

650

651 # Load the project metadata.

652 with open(DATA_DIRECTORY / project_id / "metadata.json", "r") as metadata_fileobject:

653 # Return the project metadata.

654 return {

655 "project_id": project_id,

656 "metadata": json.load(metadata_fileobject),

657 }

658

659

660###

661### ROUTE: Download a project in a zip archive.

662###

663@app.get(

664 "/api/projects/{project_id}/download",

665 tags=["Projects"],

666 response_class=FileResponse,

667 status_code=status.HTTP_200_OK,

668)

669async def download_project(

670 background_tasks: BackgroundTasks,

671 project_id: str = Path(

672 ...,

673 description="The ID of the project to download.",

674 ),

675) -> FileResponse:

676 """

677 Download a project in a zip archive.

678

679 Args:

680 background_tasks (BackgroundTasks): A background task to run after the return statement.

681 project_id (str): The ID of the project to download.

682

683 Raises:

684 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

685

686 Returns:

687 FileResponse: A zip archive of the project.

688 """

689

690 # Check project id.

691 if project_id not in (await get_projects()):

692 raise HTTPException(

693 status_code=status.HTTP_404_NOT_FOUND,

694 detail="The project with id '{project_id_str}' doesn't exist.".format(

695 project_id_str=str(project_id),

696 ),

697 )

698

699 # Define archive name.

700 archive_name: str = "archive-{project_id_str}.zip".format(project_id_str=str(project_id))

701 archive_path: pathlib.Path = DATA_DIRECTORY / project_id / archive_name

702

703 # Zip the project in an archive.

704 with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as archive_filewriter:

705 archive_filewriter.write(DATA_DIRECTORY / project_id / "metadata.json", arcname="metadata.json")

706 archive_filewriter.write(DATA_DIRECTORY / project_id / "status.json", arcname="status.json")

707 archive_filewriter.write(DATA_DIRECTORY / project_id / "texts.json", arcname="texts.json")

708 archive_filewriter.write(DATA_DIRECTORY / project_id / "constraints.json", arcname="constraints.json")

709 archive_filewriter.write(DATA_DIRECTORY / project_id / "settings.json", arcname="settings.json")

710 archive_filewriter.write(DATA_DIRECTORY / project_id / "sampling.json", arcname="sampling.json")

711 archive_filewriter.write(DATA_DIRECTORY / project_id / "clustering.json", arcname="clustering.json")

712 archive_filewriter.write(DATA_DIRECTORY / project_id / "modelization.json", arcname="modelization.json")

713 if "vectors_2D.json" in os.listdir(DATA_DIRECTORY / project_id):

714 archive_filewriter.write(DATA_DIRECTORY / project_id / "vectors_2D.json", arcname="vectors_2D.json")

715 if "vectors_3D.json" in os.listdir(DATA_DIRECTORY / project_id):

716 archive_filewriter.write(DATA_DIRECTORY / project_id / "vectors_3D.json", arcname="vectors_3D.json")

717

718 # Define a backgroundtask to clear archive after downloading.

719 def clear_after_download_project(): # noqa: WPS430 (nested function)

720 """

721 Delete the archive file.

722 """

723

724 # Delete archive file.

725 if os.path.exists(archive_path): # pragma: no cover

726 os.remove(archive_path)

727

728 # Add the background task.

729 background_tasks.add_task(

730 func=clear_after_download_project,

731 )

732

733 # Return the zip archive of the project.

734 return FileResponse(

735 archive_path,

736 media_type="application/x-zip-compressed",

737 filename=archive_name,

738 )

739

740

741###

742### ROUTE: Import a project.

743###

744@app.put(

745 "/api/projects",

746 tags=["Projects"],

747 status_code=status.HTTP_201_CREATED,

748)

749async def import_project(

750 background_tasks: BackgroundTasks,

751 project_archive: UploadFile = File(

752 ...,

753 description="A zip archive representing a project. Use format from `download` route.",

754 # TODO: max_size="8MB",

755 ),

756) -> Dict[str, Any]:

757 """

758 Import a project from a zip archive file.

759

760 Args:

761 background_tasks (BackgroundTasks): A background task to run after the return statement.

762 project_archive (UploadFile, optional): A zip archive representing a project. Use format from `download` route.

763

764 Raises:

765 HTTPException: Raises `HTTP_400_NOT_FOUND` if archive is invalid.

766

767 Returns:

768 Dict[str, Any]: A dictionary that contains the ID of the imported project.

769 """

770

771 # Check archive type.

772 if project_archive.content_type != "application/x-zip-compressed":

773 raise HTTPException(

774 status_code=status.HTTP_400_BAD_REQUEST,

775 detail="The file type '{project_archive_type}' is not supported. Please use '.zip' file.".format(

776 project_archive_type=str(project_archive.content_type),

777 ),

778 )

779

780 # Temporarly store zip archive.

781 current_timestamp: float = datetime.now().timestamp()

782 new_current_project_id: str = str(int(current_timestamp * 10**6))

783 import_archive_name: str = "import-{new_current_project_id_str}.zip".format(

784 new_current_project_id_str=str(new_current_project_id)

785 )

786 import_archive_path: pathlib.Path = DATA_DIRECTORY / import_archive_name

787 with open(import_archive_path, "wb") as import_archive_fileobject_w:

788 shutil.copyfileobj(project_archive.file, import_archive_fileobject_w)

789

790 # Define a backgroundtask to clear archive after importation.

791 def clear_after_import_project(): # noqa: WPS430 (nested function)

792 """

793 Delete the archive file.

794 """

795

796 # Delete archive file.

797 if os.path.exists(import_archive_path): # pragma: no cover

798 os.remove(import_archive_path)

799

800 # Add the background task.

801 background_tasks.add_task(

802 func=clear_after_import_project,

803 )

804

805 # Try to open archive file.

806 try:

807 with zipfile.ZipFile(import_archive_path, "r") as import_archive_file:

808 ###

809 ### Check archive content.

810 ###

811 missing_files: List[str] = [

812 needed_file

813 for needed_file in (

814 "metadata.json",

815 "status.json",

816 "texts.json",

817 "constraints.json",

818 "settings.json",

819 "sampling.json",

820 "clustering.json",

821 "modelization.json", # Will be recomputed during modelization step.

822 # "vectors_2D.json", # Will be recomputed during modelization step.

823 # "vectors_3D.json", # Will be recomputed during modelization step.

824 )

825 if needed_file not in import_archive_file.namelist()

826 ]

827 if len(missing_files) != 0: # noqa: WPS507

828 raise ValueError(

829 "The project archive file doesn't contains the following files: {missing_files_str}.".format(

830 missing_files_str=str(missing_files),

831 )

832 )

833

834 ###

835 ### Check `metadata.json`.

836 ###

837 with import_archive_file.open("metadata.json") as metadata_fileobject_r:

838 metadata: Dict[str, Any] = json.load(metadata_fileobject_r)

839 metadata["project_id"] = new_current_project_id

840 if (

841 "project_name" not in metadata.keys()

842 or not isinstance(metadata["project_name"], str)

843 or "creation_timestamp" not in metadata.keys()

844 or not isinstance(metadata["creation_timestamp"], float)

845 ):

846 raise ValueError("The project archive file has an invalid `metadata.json` file.")

847

848 ###

849 ### Check `status.json`.

850 ###

851 with import_archive_file.open("status.json") as status_fileobject_r:

852 project_status: Dict[str, Any] = json.load(status_fileobject_r)

853

854 # Check `status.state`.

855 if "state" not in project_status.keys():

856 raise ValueError("The project archive file has an invalid `status.json` file (see key `state`).")

857

858 # Force `status.state` - Case of initialization.

859 if (

860 project_status["state"] == ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION # noqa: WPS514

861 or project_status["state"] == ICGUIStates.INITIALIZATION_WITH_PENDING_MODELIZATION

862 or project_status["state"] == ICGUIStates.INITIALIZATION_WITH_WORKING_MODELIZATION

863 or project_status["state"] == ICGUIStates.INITIALIZATION_WITH_ERRORS

864 ):

865 project_status["state"] = ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION

866 # Force `status.state` - Case of sampling.

867 elif (

868 project_status["state"] == ICGUIStates.SAMPLING_TODO # noqa: WPS514

869 or project_status["state"] == ICGUIStates.SAMPLING_PENDING

870 or project_status["state"] == ICGUIStates.SAMPLING_WORKING

871 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION

872 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_PENDING_MODELIZATION

873 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_WORKING_MODELIZATION

874 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_ERRORS

875 ):

876 project_status["state"] = ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION

877 # Force `status.state` - Case of annotation.

878 elif (

879 project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION # noqa: WPS514

880 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

881 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITHOUT_CONFLICTS

882 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITHOUT_CONFLICTS

883 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

884 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITH_CONFLICTS

885 or project_status["state"] == ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITH_CONFLICTS

886 or project_status["state"] == ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITHOUT_MODELIZATION

887 or project_status["state"] == ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITH_PENDING_MODELIZATION

888 or project_status["state"] == ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITH_WORKING_MODELIZATION

889 ):

890 project_status["state"] = ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITHOUT_MODELIZATION

891 # Force `status.state` - Case of clustering.

892 elif (

893 project_status["state"] == ICGUIStates.CLUSTERING_TODO # noqa: WPS514

894 or project_status["state"] == ICGUIStates.CLUSTERING_PENDING

895 or project_status["state"] == ICGUIStates.CLUSTERING_WORKING

896 or project_status["state"] == ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITHOUT_MODELIZATION

897 or project_status["state"] == ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITH_PENDING_MODELIZATION

898 or project_status["state"] == ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITH_WORKING_MODELIZATION

899 or project_status["state"] == ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITH_ERRORS

900 ):

901 project_status["state"] = ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITHOUT_MODELIZATION

902 # Force `status.state` - Case of iteration end.

903 elif (

904 project_status["state"] == ICGUIStates.ITERATION_END # noqa: WPS514

905 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION

906 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_PENDING_MODELIZATION

907 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_WORKING_MODELIZATION

908 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_ERRORS

909 ):

910 project_status["state"] = ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION

911 # Force `state` - Case of unknown state.

912 else:

913 raise ValueError("The project archive file has an invalid `status.json` file (see key `state`).")

914

915 # Force `status.task`.

916 project_status["task"] = None

917

918 # TODO: Check `texts.json`.

919 with import_archive_file.open("texts.json") as texts_fileobject_r:

920 texts: Dict[str, Dict[str, Any]] = json.load(texts_fileobject_r)

921

922 # TODO: Check `constraints.json`.

923 with import_archive_file.open("constraints.json") as constraints_fileobject_r:

924 constraints: Dict[str, Dict[str, Any]] = json.load(constraints_fileobject_r)

925

926 # TODO: Check `settings.json`.

927 with import_archive_file.open("settings.json") as settings_fileobject_r:

928 settings: Dict[str, Dict[str, Any]] = json.load(settings_fileobject_r)

929

930 # TODO: Check `sampling.json`.

931 with import_archive_file.open("sampling.json") as sampling_fileobject_r:

932 sampling: Dict[str, List[str]] = json.load(sampling_fileobject_r)

933

934 # TODO: Check `clustering.json`.

935 with import_archive_file.open("clustering.json") as clustering_fileobject_r:

936 clustering: Dict[str, Dict[str, str]] = json.load(clustering_fileobject_r)

937

938 # TODO: Check `modelization.json`.

939 with import_archive_file.open("modelization.json") as modelization_fileobject_r:

940 modelization: Dict[str, Dict[str, Any]] = json.load(modelization_fileobject_r)

941

942 # Error: case of custom raised errors.

943 except ValueError as value_error:

944 raise HTTPException(

945 status_code=status.HTTP_400_BAD_REQUEST,

946 detail=str(value_error),

947 )

948

949 # Error: other raised errors.

950 except Exception:

951 raise HTTPException(

952 status_code=status.HTTP_400_BAD_REQUEST,

953 detail="An error occurs in project import. Project archive is probably invalid.",

954 )

955

956 # Create the directory and subdirectories of the new project.

957 os.mkdir(DATA_DIRECTORY / metadata["project_id"])

958

959 # Store `metadata.json`.

960 with open(DATA_DIRECTORY / metadata["project_id"] / "metadata.json", "w") as metadata_fileobject_w:

961 json.dump(metadata, metadata_fileobject_w, indent=4)

962

963 # Store `status.json`.

964 with open(DATA_DIRECTORY / metadata["project_id"] / "status.json", "w") as status_fileobject_w:

965 json.dump(project_status, status_fileobject_w, indent=4)

966

967 # Store `texts.json`.

968 with open(DATA_DIRECTORY / metadata["project_id"] / "texts.json", "w") as texts_fileobject_w:

969 json.dump(texts, texts_fileobject_w, indent=4)

970

971 # Store `constraints.json`.

972 with open(DATA_DIRECTORY / metadata["project_id"] / "constraints.json", "w") as constraints_fileobject_w:

973 json.dump(constraints, constraints_fileobject_w, indent=4)

974

975 # Store `settings.json`.

976 with open(DATA_DIRECTORY / metadata["project_id"] / "settings.json", "w") as settings_fileobject_w:

977 json.dump(settings, settings_fileobject_w, indent=4)

978

979 # Store `sampling.json`.

980 with open(DATA_DIRECTORY / metadata["project_id"] / "sampling.json", "w") as sampling_fileobject_w:

981 json.dump(sampling, sampling_fileobject_w, indent=4)

982

983 # Store `clustering.json`.

984 with open(DATA_DIRECTORY / metadata["project_id"] / "clustering.json", "w") as clustering_fileobject_w:

985 json.dump(clustering, clustering_fileobject_w, indent=4)

986

987 # Store `modelization.json`.

988 with open(DATA_DIRECTORY / metadata["project_id"] / "modelization.json", "w") as modelization_fileobject_w:

989 json.dump(modelization, modelization_fileobject_w, indent=4)

990

991 # Return the new ID of the imported project.

992 return {

993 "project_id": metadata["project_id"],

994 "detail": (

995 "The project with name '{project_name_str}' has been imported. It has the id '{project_id_str}'.".format(

996 project_name_str=str(metadata["project_name"]),

997 project_id_str=str(metadata["project_id"]),

998 )

999 ),

1000 }

1001

1002

1003###

1004### ROUTE: Get HTML project home page.

1005###

1006@app.get(

1007 "/gui/projects/{project_id}",

1008 tags=["Projects"],

1009 response_class=Response,

1010 status_code=status.HTTP_200_OK,

1011)

1012async def get_html_project_home_page(

1013 request: Request,

1014 project_id: str = Path(

1015 ...,

1016 description="The ID of the project.",

1017 ),

1018) -> Response:

1019 """

1020 Get HTML project home page.

1021

1022 Args:

1023 request (Request): The request context.

1024 project_id (str): The ID of the project.

1025

1026 Returns:

1027 Response: The requested page.

1028 """

1029

1030 # Return HTML project home page.

1031 try:

1032 return templates.TemplateResponse(

1033 name="project_home.html",

1034 context={

1035 "request": request,

1036 # Get the project ID.

1037 "project_id": project_id,

1038 # Get the project metadata (ID, name, creation date).

1039 "metadata": (await get_metadata(project_id=project_id))["metadata"],

1040 # Get the project status (iteration, step name and status, modelization state and conflict).

1041 "status": (await get_status(project_id=project_id))["status"],

1042 # Get the project constraints.

1043 "constraints": (

1044 await get_constraints(

1045 project_id=project_id,

1046 without_hidden_constraints=True,

1047 sorted_by=ConstraintsSortOptions.ITERATION_OF_SAMPLING,

1048 sorted_reverse=False,

1049 )

1050 )["constraints"],

1051 },

1052 status_code=status.HTTP_200_OK,

1053 )

1054

1055 # Case of error: Return HTML error page.

1056 except HTTPException as error:

1057 # Return HTML error page.

1058 return templates.TemplateResponse(

1059 name="error.html",

1060 context={

1061 "request": request,

1062 "status_code": error.status_code,

1063 "detail": error.detail,

1064 },

1065 status_code=error.status_code,

1066 )

1067

1068

1069# ==============================================================================

1070# DEFINE ROUTES FOR STATUS

1071# ==============================================================================

1072

1073

1074###

1075### ROUTE: Get status.

1076###

1077@app.get(

1078 "/api/projects/{project_id}/status",

1079 tags=["Status"],

1080 status_code=status.HTTP_200_OK,

1081)

1082async def get_status(

1083 project_id: str = Path(

1084 ...,

1085 description="The ID of the project.",

1086 ),

1087) -> Dict[str, Any]:

1088 """

1089 Get status.

1090

1091 Args:

1092 project_id (str): The ID of the project.

1093

1094 Raises:

1095 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1096

1097 Returns:

1098 Dict[str, Any]: A dictionary that contains status.

1099 """

1100

1101 # Check project id.

1102 if project_id not in (await get_projects()):

1103 raise HTTPException(

1104 status_code=status.HTTP_404_NOT_FOUND,

1105 detail="The project with id '{project_id_str}' doesn't exist.".format(

1106 project_id_str=str(project_id),

1107 ),

1108 )

1109

1110 # Load status file.

1111 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

1112 project_status: Dict[str, Any] = json.load(status_fileobject)

1113 project_status["state_details"] = get_ICGUIStates_details(state=project_status["state"])

1114

1115 # Return the requested status.

1116 return {"project_id": project_id, "status": project_status}

1117

1118

1119###

1120### ROUTE: Move to next iteration after clustering step.

1121###

1122@app.post(

1123 "/api/projects/{project_id}/iterations",

1124 tags=["Status"],

1125 status_code=status.HTTP_201_CREATED,

1126)

1127async def move_to_next_iteration(

1128 project_id: str = Path(

1129 ...,

1130 description="The ID of the project.",

1131 ),

1132) -> Dict[str, Any]:

1133 """

1134 Move to next iteration after clustering step.

1135

1136 Args:

1137 project_id (str): The ID of the project.

1138

1139 Raises:

1140 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1141 HTTPException: Raises `HTTP_403_FORBIDDEN` if the project didn't complete its clustering step.

1142

1143 Returns:

1144 Dict[str, Any]: A dictionary that contains the ID of the new iteration.

1145 """

1146

1147 # Check project id.

1148 if project_id not in (await get_projects()):

1149 raise HTTPException(

1150 status_code=status.HTTP_404_NOT_FOUND,

1151 detail="The project with id '{project_id_str}' doesn't exist.".format(

1152 project_id_str=str(project_id),

1153 ),

1154 )

1155

1156 # Lock status file in order to check project status for this step.

1157 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

1158 ###

1159 ### Load needed data.

1160 ###

1161

1162 # Load status file.

1163 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject_r:

1164 project_status: Dict[str, Any] = json.load(status_fileobject_r)

1165

1166 # Load settings file.

1167 with open(DATA_DIRECTORY / project_id / "settings.json", "r") as settings_fileobject_r:

1168 project_settings: Dict[str, Any] = json.load(settings_fileobject_r)

1169

1170 # Get current iteration id.

1171 current_iteration_id: int = project_status["iteration_id"]

1172

1173 ###

1174 ### Check parameters.

1175 ###

1176

1177 # Check project status.

1178 if project_status["state"] != ICGUIStates.ITERATION_END:

1179 raise HTTPException(

1180 status_code=status.HTTP_403_FORBIDDEN,

1181 detail="The project with id '{project_id_str}' hasn't completed its clustering step on iteration '{iteration_id_str}'.".format(

1182 project_id_str=str(project_id),

1183 iteration_id_str=str(current_iteration_id),

1184 ),

1185 )

1186

1187 ###

1188 ### Update data.

1189 ###

1190

1191 # Define new iteration id.

1192 new_iteration_id: int = current_iteration_id + 1

1193

1194 # Initialize status for the new iteration.

1195 project_status["iteration_id"] = new_iteration_id

1196 project_status["state"] = ICGUIStates.SAMPLING_TODO

1197

1198 # Initialize settings for the new iteration.

1199 project_settings[str(new_iteration_id)] = {

1200 "sampling": (

1201 default_SamplingSettingsModel().to_dict()

1202 if (current_iteration_id == 0)

1203 else project_settings[str(current_iteration_id)]["sampling"]

1204 ),

1205 "preprocessing": project_settings[str(current_iteration_id)]["preprocessing"],

1206 "vectorization": project_settings[str(current_iteration_id)]["vectorization"],

1207 "clustering": project_settings[str(current_iteration_id)]["clustering"],

1208 }

1209

1210 ###

1211 ### Store updated data.

1212 ###

1213

1214 # Store project settings.

1215 with open(DATA_DIRECTORY / project_id / "settings.json", "w") as settings_fileobject_w:

1216 json.dump(project_settings, settings_fileobject_w, indent=4)

1217

1218 # Store project status.

1219 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

1220 json.dump(project_status, status_fileobject_w, indent=4)

1221

1222 # Return the new iteration id.

1223 return {

1224 "project_id": project_id,

1225 "iteration_id": new_iteration_id,

1226 "detail": "The project with id '{project_id_str}' is now on iteration with id '{iteration_id_str}'.".format(

1227 project_id_str=str(project_id),

1228 iteration_id_str=str(new_iteration_id),

1229 ),

1230 }

1231

1232

1233# ==============================================================================

1234# DEFINE ROUTES FOR TEXTS

1235# ==============================================================================

1236

1237

1238###

1239### ROUTE: Get texts.

1240###

1241@app.get(

1242 "/api/projects/{project_id}/texts",

1243 tags=["Texts"],

1244 status_code=status.HTTP_200_OK,

1245)

1246async def get_texts(

1247 project_id: str = Path(

1248 ...,

1249 description="The ID of the project.",

1250 ),

1251 without_deleted_texts: bool = Query(

1252 True,

1253 description="The option to not return deleted texts. Defaults to `True`.",

1254 ),

1255 sorted_by: TextsSortOptions = Query(

1256 TextsSortOptions.ALPHABETICAL,

1257 description="The option to sort texts. Defaults to `ALPHABETICAL`.",

1258 ),

1259 sorted_reverse: bool = Query(

1260 False,

1261 description="The option to reverse texts order. Defaults to `False`.",

1262 ),

1263 # TODO: filter_text

1264 # TODO: limit_size + offset

1265) -> Dict[str, Any]:

1266 """

1267 Get texts.

1268

1269 Args:

1270 project_id (str): The ID of the project.

1271 without_deleted_texts (bool): The option to not return deleted texts. Defaults to `True`.

1272 sorted_by (TextsSortOptions, optional): The option to sort texts. Defaults to `ALPHABETICAL`.

1273 sorted_reverse (bool, optional): The option to reverse texts order. Defaults to `False`.

1274

1275 Raises:

1276 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1277

1278 Returns:

1279 Dict[str, Any]: A dictionary that contains texts.

1280 """

1281

1282 # Check project id.

1283 if project_id not in (await get_projects()):

1284 raise HTTPException(

1285 status_code=status.HTTP_404_NOT_FOUND,

1286 detail="The project with id '{project_id_str}' doesn't exist.".format(

1287 project_id_str=str(project_id),

1288 ),

1289 )

1290

1291 ###

1292 ### Load needed data.

1293 ###

1294

1295 # Load texts.

1296 with open(DATA_DIRECTORY / project_id / "texts.json", "r") as texts_fileobject:

1297 texts: Dict[str, Any] = {

1298 text_id: text_value

1299 for text_id, text_value in json.load(texts_fileobject).items()

1300 if (without_deleted_texts is False or text_value["is_deleted"] is False)

1301 }

1302

1303 ###

1304 ### Sort texts.

1305 ###

1306

1307 # Define the values selection method.

1308 def get_value_for_texts_sorting(text_to_sort: Tuple[str, Dict[str, Any]]) -> Any: # noqa: WPS430 (nested function)

1309 """Return the values expected for texts sorting.

1310

1311 Args:

1312 text_to_sort (Tuple[str, Dict[str, Any]]): A text (from `.items()`).

1313

1314 Returns:

1315 Any: The expected values of the text need for the sort.

1316 """

1317 # By text id.

1318 if sorted_by == TextsSortOptions.ID:

1319 return text_to_sort[0]

1320 # By text value.

1321 if sorted_by == TextsSortOptions.ALPHABETICAL:

1322 return text_to_sort[1]["text_preprocessed"]

1323 # By deletion status.

1324 #### if sorted_by == TextsSortOptions.IS_DELETED:

1325 return text_to_sort[1]["is_deleted"]

1326

1327 # Sorted the texts to return.

1328 sorted_texts: Dict[str, Any] = dict(

1329 sorted(

1330 texts.items(),

1331 key=get_value_for_texts_sorting,

1332 reverse=sorted_reverse,

1333 )

1334 )

1335

1336 # Return the requested texts.

1337 return {

1338 "project_id": project_id,

1339 "texts": sorted_texts,

1340 # Get the request parameters.

1341 "parameters": {

1342 "without_deleted_texts": without_deleted_texts,

1343 "sorted_by": sorted_by.value,

1344 "sorted_reverse": sorted_reverse,

1345 },

1346 }

1347

1348

1349###

1350### ROUTE: Delete a text.

1351###

1352@app.put(

1353 "/api/projects/{project_id}/texts/{text_id}/delete",

1354 tags=["Texts"],

1355 status_code=status.HTTP_202_ACCEPTED,

1356)

1357async def delete_text(

1358 project_id: str = Path(

1359 ...,

1360 description="The ID of the project.",

1361 ),

1362 text_id: str = Path(

1363 ...,

1364 description="The ID of the text.",

1365 ),

1366) -> Dict[str, Any]:

1367 """

1368 Delete a text.

1369

1370 Args:

1371 project_id (str): The ID of the project.

1372 text_id (str): The ID of the text.

1373

1374 Raises:

1375 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1376 HTTPException: Raises `HTTP_404_NOT_FOUND` if the text with id `text_id` to delete doesn't exist.

1377 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow modification.

1378

1379 Returns:

1380 Dict[str, Any]: A dictionary that contains the ID of deleted text.

1381 """

1382

1383 # Check project id.

1384 if project_id not in (await get_projects()):

1385 raise HTTPException(

1386 status_code=status.HTTP_404_NOT_FOUND,

1387 detail="The project with id '{project_id_str}' doesn't exist.".format(

1388 project_id_str=str(project_id),

1389 ),

1390 )

1391

1392 # Lock status file in order to check project status for this step.

1393 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

1394 ###

1395 ### Load needed data.

1396 ###

1397

1398 # Load status file.

1399 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

1400 project_status: Dict[str, Any] = json.load(status_fileobject)

1401

1402 # Load texts file.

1403 with open(DATA_DIRECTORY / project_id / "texts.json", "r") as texts_fileobject_r:

1404 texts: Dict[str, Any] = json.load(texts_fileobject_r)

1405

1406 # Load constraints file.

1407 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject_r:

1408 constraints: Dict[str, Any] = json.load(constraints_fileobject_r)

1409

1410 ###

1411 ### Check parameters.

1412 ###

1413

1414 # Check text id.

1415 if text_id not in texts.keys():

1416 raise HTTPException(

1417 status_code=status.HTTP_404_NOT_FOUND,

1418 detail="In project with id '{project_id_str}', the text with id '{text_id_str}' to delete doesn't exist.".format(

1419 project_id_str=str(project_id),

1420 text_id_str=str(text_id),

1421 ),

1422 )

1423

1424 # Check status.

1425 if (

1426 project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION # noqa: WPS514

1427 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1428 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1429 ):

1430 raise HTTPException(

1431 status_code=status.HTTP_403_FORBIDDEN,

1432 detail="The project with id '{project_id_str}' doesn't allow modification during this state (state='{state_str}').".format(

1433 project_id_str=str(project_id),

1434 state_str=str(project_status["state"]),

1435 ),

1436 )

1437

1438 ###

1439 ### Update data.

1440 ###

1441

1442 # Update status by forcing "outdated" status.

1443 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

1444 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1445 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

1446 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1447 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

1448 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1449

1450 # Update texts by deleting the text.

1451 texts[text_id]["is_deleted"] = True

1452

1453 # Update constraints by hidding those associated with the deleted text.

1454 for constraint_id, constraint_value in constraints.items():

1455 data_id1: str = constraint_value["data"]["id_1"]

1456 data_id2: str = constraint_value["data"]["id_2"]

1457

1458 if text_id in {

1459 data_id1,

1460 data_id2,

1461 }:

1462 constraints[constraint_id]["is_hidden"] = True

1463

1464 ###

1465 ### Store updated data.

1466 ###

1467

1468 # Store updated status in file.

1469 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

1470 json.dump(project_status, status_fileobject_w, indent=4)

1471

1472 # Store updated texts in file.

1473 with open(DATA_DIRECTORY / project_id / "texts.json", "w") as texts_fileobject_w:

1474 json.dump(texts, texts_fileobject_w, indent=4)

1475

1476 # Store updated constraints in file.

1477 with open(DATA_DIRECTORY / project_id / "constraints.json", "w") as constraints_fileobject_w:

1478 json.dump(constraints, constraints_fileobject_w, indent=4)

1479

1480 # Return statement.

1481 return {

1482 "project_id": project_id,

1483 "text_id": text_id,

1484 "detail": "In project with id '{project_id_str}', the text with id '{text_id_str}' has been deleted. Several constraints have been hidden.".format(

1485 project_id_str=str(project_id),

1486 text_id_str=str(text_id),

1487 ),

1488 }

1489

1490

1491###

1492### ROUTE: Undelete a text.

1493###

1494@app.put(

1495 "/api/projects/{project_id}/texts/{text_id}/undelete",

1496 tags=["Texts"],

1497 status_code=status.HTTP_202_ACCEPTED,

1498)

1499async def undelete_text(

1500 project_id: str = Path(

1501 ...,

1502 description="The ID of the project.",

1503 ),

1504 text_id: str = Path(

1505 ...,

1506 description="The ID of the text.",

1507 ),

1508) -> Dict[str, Any]:

1509 """

1510 Undelete a text.

1511

1512 Args:

1513 project_id (str): The ID of the project.

1514 text_id (str): The ID of the text.

1515

1516 Raises:

1517 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1518 HTTPException: Raises `HTTP_404_NOT_FOUND` if the text with id `text_id` to undelete doesn't exist.

1519 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow modification.

1520

1521 Returns:

1522 Dict[str, Any]: A dictionary that contains the ID of undeleted text.

1523 """

1524

1525 # Check project id.

1526 if project_id not in (await get_projects()):

1527 raise HTTPException(

1528 status_code=status.HTTP_404_NOT_FOUND,

1529 detail="The project with id '{project_id_str}' doesn't exist.".format(

1530 project_id_str=str(project_id),

1531 ),

1532 )

1533

1534 # Lock status file in order to check project status for this step.

1535 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

1536 ###

1537 ### Load needed data.

1538 ###

1539

1540 # Load status file.

1541 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

1542 project_status: Dict[str, Any] = json.load(status_fileobject)

1543

1544 # Load texts file.

1545 with open(DATA_DIRECTORY / project_id / "texts.json", "r") as texts_fileobject_r:

1546 texts: Dict[str, Any] = json.load(texts_fileobject_r)

1547

1548 # Load constraints file.

1549 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject_r:

1550 constraints: Dict[str, Any] = json.load(constraints_fileobject_r)

1551

1552 ###

1553 ### Check parameters.

1554 ###

1555

1556 # Check text id.

1557 if text_id not in texts.keys():

1558 raise HTTPException(

1559 status_code=status.HTTP_404_NOT_FOUND,

1560 detail="In project with id '{project_id_str}', the text with id '{text_id_str}' to undelete doesn't exist.".format(

1561 project_id_str=str(project_id),

1562 text_id_str=str(text_id),

1563 ),

1564 )

1565

1566 # Check status.

1567 if (

1568 project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION # noqa: WPS514

1569 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1570 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1571 ):

1572 raise HTTPException(

1573 status_code=status.HTTP_403_FORBIDDEN,

1574 detail="The project with id '{project_id_str}' doesn't allow modification during this state (state='{state_str}').".format(

1575 project_id_str=str(project_id),

1576 state_str=str(project_status["state"]),

1577 ),

1578 )

1579

1580 ###

1581 ### Update data.

1582 ###

1583

1584 # Update status by forcing "outdated" status.

1585 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

1586 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1587 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

1588 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1589 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

1590 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1591

1592 # Update texts by undeleting the text.

1593 texts[text_id]["is_deleted"] = False

1594

1595 # Update constraints by unhidding those associated with the undeleted text.

1596 for constraint_id, constraint_value in constraints.items():

1597 data_id1: str = constraint_value["data"]["id_1"]

1598 data_id2: str = constraint_value["data"]["id_2"]

1599

1600 if text_id in {data_id1, data_id2}:

1601 constraints[constraint_id]["is_hidden"] = (

1602 texts[data_id1]["is_deleted"] is True or texts[data_id2]["is_deleted"] is True

1603 )

1604

1605 ###

1606 ### Store updated data.

1607 ###

1608

1609 # Store updated status in file.

1610 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

1611 json.dump(project_status, status_fileobject_w, indent=4)

1612

1613 # Store updated texts in file.

1614 with open(DATA_DIRECTORY / project_id / "texts.json", "w") as texts_fileobject_w:

1615 json.dump(texts, texts_fileobject_w, indent=4)

1616

1617 # Store updated constraints in file.

1618 with open(DATA_DIRECTORY / project_id / "constraints.json", "w") as constraints_fileobject_w:

1619 json.dump(constraints, constraints_fileobject_w, indent=4)

1620

1621 # Return statement.

1622 return {

1623 "project_id": project_id,

1624 "text_id": text_id,

1625 "detail": "In project with id '{project_id_str}', the text with id '{text_id_str}' has been undeleted. Several constraints have been unhidden.".format(

1626 project_id_str=str(project_id),

1627 text_id_str=str(text_id),

1628 ),

1629 }

1630

1631

1632###

1633### ROUTE: Rename a text.

1634###

1635@app.put(

1636 "/api/projects/{project_id}/texts/{text_id}/rename",

1637 tags=["Texts"],

1638 status_code=status.HTTP_202_ACCEPTED,

1639)

1640async def rename_text(

1641 project_id: str = Path(

1642 ...,

1643 description="The ID of the project.",

1644 ),

1645 text_id: str = Path(

1646 ...,

1647 description="The ID of the text.",

1648 ),

1649 text_value: str = Query(

1650 ...,

1651 description="The new value of the text.",

1652 min_length=3,

1653 max_length=256,

1654 ),

1655) -> Dict[str, Any]:

1656 """

1657 Rename a text.

1658

1659 Args:

1660 project_id (str): The ID of the project.

1661 text_id (str): The ID of the text.

1662 text_value (str): The new value of the text.

1663

1664 Raises:

1665 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1666 HTTPException: Raises `HTTP_404_NOT_FOUND` if the text with id `text_id` to rename doesn't exist.

1667 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow modification.

1668

1669 Returns:

1670 Dict[str, Any]: A dictionary that contains the ID of renamed text.

1671 """

1672

1673 # Check project id.

1674 if project_id not in (await get_projects()):

1675 raise HTTPException(

1676 status_code=status.HTTP_404_NOT_FOUND,

1677 detail="The project with id '{project_id_str}' doesn't exist.".format(

1678 project_id_str=str(project_id),

1679 ),

1680 )

1681

1682 # Lock status file in order to check project status for this step.

1683 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

1684 ###

1685 ### Load needed data.

1686 ###

1687

1688 # Load status file.

1689 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

1690 project_status: Dict[str, Any] = json.load(status_fileobject)

1691

1692 # Load texts file.

1693 with open(DATA_DIRECTORY / project_id / "texts.json", "r") as texts_fileobject_r:

1694 texts: Dict[str, Any] = json.load(texts_fileobject_r)

1695

1696 ###

1697 ### Check parameters.

1698 ###

1699

1700 # Check text id.

1701 if text_id not in texts.keys():

1702 raise HTTPException(

1703 status_code=status.HTTP_404_NOT_FOUND,

1704 detail="In project with id '{project_id_str}', the text with id '{text_id_str}' to rename doesn't exist.".format(

1705 project_id_str=str(project_id),

1706 text_id_str=str(text_id),

1707 ),

1708 )

1709

1710 # Check status.

1711 if (

1712 project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION # noqa: WPS514

1713 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1714 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1715 ):

1716 raise HTTPException(

1717 status_code=status.HTTP_403_FORBIDDEN,

1718 detail="The project with id '{project_id_str}' doesn't allow modification during this state (state='{state_str}').".format(

1719 project_id_str=str(project_id),

1720 state_str=str(project_status["state"]),

1721 ),

1722 )

1723

1724 ###

1725 ### Update data.

1726 ###

1727

1728 # Update status by forcing "outdated" status.

1729 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

1730 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1731 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

1732 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

1733 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

1734 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

1735

1736 # Update texts by renaming the new text.

1737 texts[text_id]["text"] = text_value

1738

1739 ###

1740 ### Store updated data.

1741 ###

1742

1743 # Store updated status in file.

1744 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

1745 json.dump(project_status, status_fileobject_w, indent=4)

1746

1747 # Store updated texts in file.

1748 with open(DATA_DIRECTORY / project_id / "texts.json", "w") as texts_fileobject_w:

1749 json.dump(texts, texts_fileobject_w, indent=4)

1750

1751 # Return statement.

1752 return {

1753 "project_id": project_id,

1754 "text_id": text_id,

1755 "text_value": text_value,

1756 "detail": "In project with id '{project_id_str}', the text with id '{text_id_str}' has been renamed.".format(

1757 project_id_str=str(project_id),

1758 text_id_str=str(text_id),

1759 ),

1760 }

1761

1762

1763###

1764### ROUTE: Get HTML texts page.

1765###

1766@app.get(

1767 "/gui/projects/{project_id}/texts",

1768 tags=["Texts"],

1769 response_class=Response,

1770 status_code=status.HTTP_200_OK,

1771)

1772async def get_html_texts_page(

1773 request: Request,

1774 project_id: str = Path(

1775 ...,

1776 description="The ID of the project.",

1777 ),

1778 sorted_by: TextsSortOptions = Query(

1779 TextsSortOptions.ALPHABETICAL,

1780 description="The option to sort texts. Defaults to `ALPHABETICAL`.",

1781 ),

1782 sorted_reverse: bool = Query(

1783 False,

1784 description="The option to reverse texts order. Defaults to `False`.",

1785 ),

1786 # TODO: filter_text

1787 # TODO: limit_size + offset

1788) -> Response:

1789 """

1790 Get HTML texts page.

1791

1792 Args:

1793 request (Request): The request context.

1794 project_id (str): The ID of the project.

1795 sorted_by (TextsSortOptions, optional): The option to sort texts. Defaults to `ALPHABETICAL`.

1796 sorted_reverse (bool, optional): The option to reverse texts order. Defaults to `False`.

1797

1798 Returns:

1799 Response: The requested page.

1800 """

1801

1802 # Return HTML constraints page.

1803 try:

1804 return templates.TemplateResponse(

1805 name="texts.html",

1806 context={

1807 "request": request,

1808 # Get the project ID.

1809 "project_id": project_id,

1810 # Get the request parameters.

1811 "parameters": {

1812 "without_deleted_texts": True,

1813 "sorted_by": sorted_by.value,

1814 "sorted_reverse": sorted_reverse,

1815 },

1816 # Get the project metadata (ID, name, creation date).

1817 "metadata": (await get_metadata(project_id=project_id))["metadata"],

1818 # Get the project status (iteration, step name and status, modelization state and conflict).

1819 "status": (await get_status(project_id=project_id))["status"],

1820 # Get the project texts.

1821 "texts": (

1822 await get_texts(

1823 project_id=project_id,

1824 without_deleted_texts=False,

1825 sorted_by=sorted_by,

1826 sorted_reverse=sorted_reverse,

1827 )

1828 )["texts"],

1829 # Get the project constraints.

1830 "constraints": (

1831 await get_constraints(

1832 project_id=project_id,

1833 without_hidden_constraints=True,

1834 sorted_by=ConstraintsSortOptions.ID,

1835 sorted_reverse=False,

1836 )

1837 )["constraints"],

1838 },

1839 status_code=status.HTTP_200_OK,

1840 )

1841

1842 # Case of error: Return HTML error page.

1843 except HTTPException as error:

1844 # Return HTML error page.

1845 return templates.TemplateResponse(

1846 name="error.html",

1847 context={

1848 "request": request,

1849 "status_code": error.status_code,

1850 "detail": error.detail,

1851 },

1852 status_code=error.status_code,

1853 )

1854

1855

1856# ==============================================================================

1857# DEFINE ROUTES FOR CONSTRAINTS

1858# ==============================================================================

1859

1860

1861###

1862### ROUTE: Get constraints.

1863###

1864@app.get(

1865 "/api/projects/{project_id}/constraints",

1866 tags=["Constraints"],

1867 status_code=status.HTTP_200_OK,

1868)

1869async def get_constraints(

1870 project_id: str = Path(

1871 ...,

1872 description="The ID of the project.",

1873 ),

1874 without_hidden_constraints: bool = Query(

1875 True,

1876 description="The option to not return hidden constraints. Defaults to `True`.",

1877 ),

1878 sorted_by: ConstraintsSortOptions = Query(

1879 ConstraintsSortOptions.ID,

1880 description="The option to sort constraints. Defaults to `ID`.",

1881 ),

1882 sorted_reverse: bool = Query(

1883 False,

1884 description="The option to reverse constraints order. Defaults to `False`.",

1885 ),

1886 # TODO: filter_text

1887 # TODO: limit_size + offset

1888) -> Dict[str, Any]:

1889 """

1890 Get constraints.

1891

1892 Args:

1893 project_id (str): The ID of the project.

1894 without_hidden_constraints (bool, optional): The option to not return hidden constraints. Defaults to `True`.

1895 sorted_by (ConstraintsSortOptions, optional): The option to sort constraints. Defaults to `ID`.

1896 sorted_reverse (bool, optional): The option to reverse constraints order. Defaults to `False`.

1897

1898 Raises:

1899 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

1900

1901 Returns:

1902 Dict[str, Any]: A dictionary that contains constraints.

1903 """

1904

1905 # Check project id.

1906 if project_id not in (await get_projects()):

1907 raise HTTPException(

1908 status_code=status.HTTP_404_NOT_FOUND,

1909 detail="The project with id '{project_id_str}' doesn't exist.".format(

1910 project_id_str=str(project_id),

1911 ),

1912 )

1913

1914 ###

1915 ### Load needed data.

1916 ###

1917

1918 # Load constraints.

1919 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject:

1920 constraints: Dict[str, Any] = {

1921 constraint_id: constraint_value

1922 for constraint_id, constraint_value in json.load(constraints_fileobject).items()

1923 if (without_hidden_constraints is False or constraint_value["is_hidden"] is False)

1924 }

1925

1926 # Load texts.

1927 with open(DATA_DIRECTORY / project_id / "texts.json", "r") as texts_fileobject:

1928 texts: Dict[str, Any] = json.load(texts_fileobject)

1929

1930 ###

1931 ### Sort constraints.

1932 ###

1933

1934 # Define the values selection method.

1935 def get_value_for_constraints_sorting( # noqa: WPS430 (nested function)

1936 constraint_to_sort: Tuple[str, Dict[str, Any]]

1937 ) -> Any:

1938 """Return the values expected for constraints sorting.

1939

1940 Args:

1941 constraint_to_sort (Tuple[str, Dict[str, Any]]): A constraint (from `.items()`).

1942

1943 Returns:

1944 Any: The expected values of the constraint need for the sort.

1945 """

1946 # By constraint id.

1947 if sorted_by == ConstraintsSortOptions.ID:

1948 return constraint_to_sort[0]

1949 # By texts.

1950 if sorted_by == ConstraintsSortOptions.TEXT:

1951 return (

1952 texts[constraint_to_sort[1]["data"]["id_1"]]["text"],

1953 texts[constraint_to_sort[1]["data"]["id_2"]]["text"],

1954 )

1955 # By constraint type.

1956 if sorted_by == ConstraintsSortOptions.CONSTRAINT_TYPE:

1957 return (

1958 constraint_to_sort[1]["constraint_type"] is None,

1959 constraint_to_sort[1]["constraint_type"] == "CANNOT_LINK",

1960 constraint_to_sort[1]["constraint_type"] == "MUST_LINK",

1961 )

1962 # By date of update.

1963 if sorted_by == ConstraintsSortOptions.DATE_OF_UPDATE:

1964 return constraint_to_sort[1]["date_of_update"] if constraint_to_sort[1]["date_of_update"] is not None else 0

1965 # By iteration of sampling.

1966 if sorted_by == ConstraintsSortOptions.ITERATION_OF_SAMPLING:

1967 return constraint_to_sort[1]["iteration_of_sampling"]

1968 # To annotation.

1969 if sorted_by == ConstraintsSortOptions.TO_ANNOTATE:

1970 return constraint_to_sort[1]["to_annotate"] is False

1971 # To review.

1972 if sorted_by == ConstraintsSortOptions.TO_REVIEW:

1973 return constraint_to_sort[1]["to_review"] is False

1974 # To fix conflict.

1975 #### if sorted_by == ConstraintsSortOptions.TO_FIX_CONFLICT:

1976 return constraint_to_sort[1]["to_fix_conflict"] is False

1977

1978 # Sorted the constraints to return.

1979 sorted_constraints: Dict[str, Any] = dict(

1980 sorted(

1981 constraints.items(),

1982 key=get_value_for_constraints_sorting,

1983 reverse=sorted_reverse,

1984 )

1985 )

1986

1987 # Return the requested constraints.

1988 return {

1989 "project_id": project_id,

1990 "constraints": sorted_constraints,

1991 # Get the request parameters.

1992 "parameters": {

1993 "without_hidden_constraints": without_hidden_constraints,

1994 "sorted_by": sorted_by.value,

1995 "sorted_reverse": sorted_reverse,

1996 },

1997 }

1998

1999

2000###

2001### ROUTE: Annotate a constraint.

2002###

2003@app.put(

2004 "/api/projects/{project_id}/constraints/{constraint_id}/annotate",

2005 tags=["Constraints"],

2006 status_code=status.HTTP_202_ACCEPTED,

2007)

2008async def annotate_constraint(

2009 project_id: str = Path(

2010 ...,

2011 description="The ID of the project.",

2012 ),

2013 constraint_id: str = Path(

2014 ...,

2015 description="The ID of the constraint.",

2016 ),

2017 constraint_type: Optional[ConstraintsValues] = Query(

2018 None,

2019 description="The type of constraint to annotate. Defaults to `None`.",

2020 ),

2021) -> Dict[str, Any]:

2022 """

2023 Annotate a constraint.

2024

2025 Args:

2026 project_id (str): The ID of the project.

2027 constraint_id (str): The ID of the constraint.

2028 constraint_type (Optional[ConstraintsValues]): The type of constraint to annotate. Defaults to `None`.

2029

2030 Raises:

2031 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2032 HTTPException: Raises `HTTP_404_NOT_FOUND` if the constraint with id `constraint_id` to annotate doesn't exist.

2033 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow modification.

2034

2035 Returns:

2036 Dict[str, Any]: A dictionary that contains the ID of annotated constraint.

2037 """

2038

2039 # Check project id.

2040 if project_id not in (await get_projects()):

2041 raise HTTPException(

2042 status_code=status.HTTP_404_NOT_FOUND,

2043 detail="The project with id '{project_id_str}' doesn't exist.".format(

2044 project_id_str=str(project_id),

2045 ),

2046 )

2047

2048 # Lock status file in order to check project status for this step.

2049 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

2050 ###

2051 ### Load needed data.

2052 ###

2053

2054 # Load status file.

2055 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

2056 project_status: Dict[str, Any] = json.load(status_fileobject)

2057

2058 # Load constraints file.

2059 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject_r:

2060 constraints: Dict[str, Any] = json.load(constraints_fileobject_r)

2061

2062 ###

2063 ### Check parameters.

2064 ###

2065

2066 # Check constraint id.

2067 if constraint_id not in constraints.keys():

2068 raise HTTPException(

2069 status_code=status.HTTP_404_NOT_FOUND,

2070 detail="In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' to annotate doesn't exist.".format(

2071 project_id_str=str(project_id),

2072 constraint_id_str=str(constraint_id),

2073 ),

2074 )

2075

2076 # Check status.

2077 if (

2078 project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION # noqa: WPS514

2079 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2080 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2081 ):

2082 raise HTTPException(

2083 status_code=status.HTTP_403_FORBIDDEN,

2084 detail="The project with id '{project_id_str}' doesn't allow modification during this state (state='{state_str}').".format(

2085 project_id_str=str(project_id),

2086 state_str=str(project_status["state"]),

2087 ),

2088 )

2089

2090 ###

2091 ### Update data.

2092 ###

2093

2094 # Update status by forcing "outdated" status.

2095 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

2096 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2097 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

2098 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2099 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

2100 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2101

2102 # Update constraints by updating the constraint history.

2103 constraints[constraint_id]["constraint_type_previous"].append(constraints[constraint_id]["constraint_type"])

2104

2105 # Update constraints by annotating the constraint.

2106 constraints[constraint_id]["constraint_type"] = constraint_type

2107 constraints[constraint_id]["date_of_update"] = datetime.now().timestamp()

2108

2109 # Force annotation status.

2110 constraints[constraint_id]["to_annotate"] = False

2111

2112 ###

2113 ### Store updated data.

2114 ###

2115

2116 # Store updated status in file.

2117 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

2118 json.dump(project_status, status_fileobject_w, indent=4)

2119

2120 # Store updated constraints in file.

2121 with open(DATA_DIRECTORY / project_id / "constraints.json", "w") as constraints_fileobject_w:

2122 json.dump(constraints, constraints_fileobject_w, indent=4)

2123

2124 # Return statement.

2125 return {

2126 "project_id": project_id,

2127 "constraint_id": constraint_id,

2128 "detail": "In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' has been annotated at `{constraint_type_str}`.".format(

2129 project_id_str=str(project_id),

2130 constraint_id_str=str(constraint_id),

2131 constraint_type_str="None" if (constraint_type is None) else str(constraint_type.value),

2132 ),

2133 }

2134

2135

2136###

2137### ROUTE: Review a constraint.

2138###

2139@app.put(

2140 "/api/projects/{project_id}/constraints/{constraint_id}/review",

2141 tags=["Constraints"],

2142 status_code=status.HTTP_202_ACCEPTED,

2143)

2144async def review_constraint(

2145 project_id: str = Path(

2146 ...,

2147 description="The ID of the project.",

2148 ),

2149 constraint_id: str = Path(

2150 ...,

2151 description="The ID of the constraint.",

2152 ),

2153 to_review: bool = Query(

2154 True,

2155 description="The choice to review or not the constraint. Defaults to `True`.",

2156 ),

2157) -> Dict[str, Any]:

2158 """

2159 Review a constraint.

2160

2161 Args:

2162 project_id (str): The ID of the project.

2163 constraint_id (str): The ID of the constraint.

2164 to_review (str): The choice to review or not the constraint. Defaults to `True`.

2165

2166 Raises:

2167 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2168 HTTPException: Raises `HTTP_404_NOT_FOUND` if the constraint with id `constraint_id` to annotate doesn't exist.

2169

2170 Returns:

2171 Dict[str, Any]: A dictionary that contains the ID of reviewed constraint.

2172 """

2173

2174 # Check project id.

2175 if project_id not in (await get_projects()):

2176 raise HTTPException(

2177 status_code=status.HTTP_404_NOT_FOUND,

2178 detail="The project with id '{project_id_str}' doesn't exist.".format(

2179 project_id_str=str(project_id),

2180 ),

2181 )

2182

2183 # Lock status file in order to check project status for this step.

2184 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

2185 ###

2186 ### Load needed data.

2187 ###

2188

2189 # Load constraints file.

2190 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject_r:

2191 constraints: Dict[str, Any] = json.load(constraints_fileobject_r)

2192

2193 ###

2194 ### Check parameters.

2195 ###

2196

2197 # Check constraint id.

2198 if constraint_id not in constraints.keys():

2199 raise HTTPException(

2200 status_code=status.HTTP_404_NOT_FOUND,

2201 detail="In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' to annotate doesn't exist.".format(

2202 project_id_str=str(project_id),

2203 constraint_id_str=str(constraint_id),

2204 ),

2205 )

2206

2207 ###

2208 ### Update data.

2209 ###

2210

2211 # Update constraints by reviewing the constraint.

2212 constraints[constraint_id]["to_review"] = to_review

2213

2214 ###

2215 ### Store updated data.

2216 ###

2217

2218 # Store updated constraints in file.

2219 with open(DATA_DIRECTORY / project_id / "constraints.json", "w") as constraints_fileobject_w:

2220 json.dump(constraints, constraints_fileobject_w, indent=4)

2221

2222 # Return statement.

2223 return {

2224 "project_id": project_id,

2225 "constraint_id": constraint_id,

2226 "detail": "In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' {review_conclusion}.".format(

2227 project_id_str=str(project_id),

2228 constraint_id_str=str(constraint_id),

2229 review_conclusion="need a review" if (to_review) else "has been reviewed",

2230 ),

2231 }

2232

2233

2234###

2235### ROUTE: Comment a constraint.

2236###

2237@app.put(

2238 "/api/projects/{project_id}/constraints/{constraint_id}/comment",

2239 tags=["Constraints"],

2240 status_code=status.HTTP_202_ACCEPTED,

2241)

2242async def comment_constraint(

2243 project_id: str = Path(

2244 ...,

2245 description="The ID of the project.",

2246 ),

2247 constraint_id: str = Path(

2248 ...,

2249 description="The ID of the constraint.",

2250 ),

2251 constraint_comment: str = Query(

2252 ...,

2253 description="The comment of constraint.",

2254 # min_length=0,

2255 max_length=256,

2256 ),

2257) -> Dict[str, Any]:

2258 """

2259 Comment a constraint.

2260

2261 Args:

2262 project_id (str): The ID of the project.

2263 constraint_id (str): The ID of the constraint.

2264 constraint_comment (str): The comment of constraint.

2265

2266 Raises:

2267 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2268 HTTPException: Raises `HTTP_404_NOT_FOUND` if the constraint with id `constraint_id` to annotate doesn't exist.

2269

2270 Returns:

2271 Dict[str, Any]: A dictionary that contains the ID of commented constraint.

2272 """

2273

2274 # Check project id.

2275 if project_id not in (await get_projects()):

2276 raise HTTPException(

2277 status_code=status.HTTP_404_NOT_FOUND,

2278 detail="The project with id '{project_id_str}' doesn't exist.".format(

2279 project_id_str=str(project_id),

2280 ),

2281 )

2282

2283 # Lock status file in order to check project status for this step.

2284 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

2285 ###

2286 ### Load needed data.

2287 ###

2288

2289 # Load constraints file.

2290 with open(DATA_DIRECTORY / project_id / "constraints.json", "r") as constraints_fileobject_r:

2291 constraints: Dict[str, Any] = json.load(constraints_fileobject_r)

2292

2293 ###

2294 ### Check parameters.

2295 ###

2296

2297 # Check constraint id.

2298 if constraint_id not in constraints.keys():

2299 raise HTTPException(

2300 status_code=status.HTTP_404_NOT_FOUND,

2301 detail="In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' to annotate doesn't exist.".format(

2302 project_id_str=str(project_id),

2303 constraint_id_str=str(constraint_id),

2304 ),

2305 )

2306

2307 ###

2308 ### Update data.

2309 ###

2310

2311 # Update constraints by commenting the constraint.

2312 constraints[constraint_id]["comment"] = constraint_comment

2313

2314 ###

2315 ### Store updated data.

2316 ###

2317

2318 # Store updated constraints in file.

2319 with open(DATA_DIRECTORY / project_id / "constraints.json", "w") as constraints_fileobject_w:

2320 json.dump(constraints, constraints_fileobject_w, indent=4)

2321

2322 # Return statement.

2323 return {

2324 "project_id": project_id,

2325 "constraint_id": constraint_id,

2326 "constraint_comment": constraint_comment,

2327 "detail": "In project with id '{project_id_str}', the constraint with id '{constraint_id_str}' has been commented.".format(

2328 project_id_str=str(project_id),

2329 constraint_id_str=str(constraint_id),

2330 ),

2331 }

2332

2333

2334###

2335### ROUTE: Approve all constraints.

2336###

2337@app.post(

2338 "/api/projects/{project_id}/constraints/approve",

2339 tags=["Constraints"],

2340 status_code=status.HTTP_201_CREATED,

2341)

2342async def approve_all_constraints(

2343 project_id: str = Path(

2344 ...,

2345 description="The ID of the project.",

2346 ),

2347) -> Dict[str, Any]:

2348 """

2349 Approve all constraints.

2350

2351 Args:

2352 project_id (str): The ID of the project.

2353

2354 Raises:

2355 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2356 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow constraints approbation.

2357

2358 Returns:

2359 Dict[str, Any]: A dictionary that contains the confirmation of constraints approbation.

2360 """

2361

2362 # Check project id.

2363 if project_id not in (await get_projects()):

2364 raise HTTPException(

2365 status_code=status.HTTP_404_NOT_FOUND,

2366 detail="The project with id '{project_id_str}' doesn't exist.".format(

2367 project_id_str=str(project_id),

2368 ),

2369 )

2370

2371 # Lock status file in order to check project status for this step.

2372 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

2373 # Load status file.

2374 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

2375 project_status: Dict[str, Any] = json.load(status_fileobject)

2376

2377 # Check status.

2378 if project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

2379 raise HTTPException(

2380 status_code=status.HTTP_403_FORBIDDEN,

2381 detail="The project with id '{project_id_str}' doesn't allow constraints approbation during this state (state='{state_str}').".format(

2382 project_id_str=str(project_id),

2383 state_str=str(project_status["state"]),

2384 ),

2385 )

2386

2387 ###

2388 ### Update data.

2389 ###

2390

2391 # Update status to clustering step.

2392 project_status["state"] = ICGUIStates.CLUSTERING_TODO

2393

2394 ###

2395 ### Store updated data.

2396 ###

2397

2398 # Store updated status in file.

2399 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

2400 json.dump(project_status, status_fileobject_w, indent=4)

2401

2402 # Return statement.

2403 return {

2404 "project_id": project_id,

2405 "detail": "In project with id '{project_id_str}', the constraints have been approved.".format(

2406 project_id_str=str(project_id),

2407 ),

2408 }

2409

2410

2411###

2412### ROUTE: Get HTML constraints page.

2413###

2414@app.get(

2415 "/gui/projects/{project_id}/constraints",

2416 tags=["Constraints"],

2417 response_class=Response,

2418 status_code=status.HTTP_200_OK,

2419)

2420async def get_html_constraints_page(

2421 request: Request,

2422 project_id: str = Path(

2423 ...,

2424 description="The ID of the project.",

2425 ),

2426 sorted_by: ConstraintsSortOptions = Query(

2427 ConstraintsSortOptions.ITERATION_OF_SAMPLING,

2428 description="The option to sort constraints. Defaults to `ITERATION_OF_SAMPLING`.",

2429 ),

2430 sorted_reverse: bool = Query(

2431 False,

2432 description="The option to reverse constraints order. Defaults to `False`.",

2433 ),

2434 # TODO: filter_text

2435 # TODO: limit_size + offset

2436) -> Response:

2437 """

2438 Get HTML constraints page.

2439

2440 Args:

2441 request (Request): The request context.

2442 project_id (str): The ID of the project.

2443 sorted_by (ConstraintsSortOptions, optional): The option to sort constraints. Defaults to `ITERATION_OF_SAMPLING`.

2444 sorted_reverse (bool, optional): The option to reverse constraints order. Defaults to `False`.

2445

2446 Returns:

2447 Response: The requested page.

2448 """

2449

2450 # Return HTML constraints page.

2451 try:

2452 return templates.TemplateResponse(

2453 name="constraints.html",

2454 context={

2455 "request": request,

2456 # Get the project ID.

2457 "project_id": project_id,

2458 # Get the request parameters.

2459 "parameters": {

2460 "without_hidden_constraints": True,

2461 "sorted_by": sorted_by.value,

2462 "sorted_reverse": sorted_reverse,

2463 },

2464 # Get the project metadata (ID, name, creation date).

2465 "metadata": (await get_metadata(project_id=project_id))["metadata"],

2466 # Get the project status (iteration, step name and status, modelization state and conflict).

2467 "status": (await get_status(project_id=project_id))["status"],

2468 # Get the project texts.

2469 "texts": (

2470 await get_texts(

2471 project_id=project_id,

2472 without_deleted_texts=False,

2473 sorted_by=TextsSortOptions.ID,

2474 sorted_reverse=False,

2475 )

2476 )["texts"],

2477 # Get the project constraints.

2478 "constraints": (

2479 await get_constraints(

2480 project_id=project_id,

2481 without_hidden_constraints=True,

2482 sorted_by=sorted_by,

2483 sorted_reverse=sorted_reverse,

2484 )

2485 )["constraints"],

2486 },

2487 status_code=status.HTTP_200_OK,

2488 )

2489

2490 # Case of error: Return HTML error page.

2491 except HTTPException as error:

2492 # Return HTML error page.

2493 return templates.TemplateResponse(

2494 name="error.html",

2495 context={

2496 "request": request,

2497 "status_code": error.status_code,

2498 "detail": error.detail,

2499 },

2500 status_code=error.status_code,

2501 )

2502

2503

2504###

2505### ROUTE: Get HTML constraint annotation page.

2506###

2507@app.get(

2508 "/gui/projects/{project_id}/constraints/{constraint_id}",

2509 tags=["Constraints"],

2510 response_class=Response,

2511 status_code=status.HTTP_200_OK,

2512)

2513async def get_html_constraint_annotation_page(

2514 request: Request,

2515 project_id: str = Path(

2516 ...,

2517 description="The ID of the project.",

2518 ),

2519 constraint_id: str = Path(

2520 ...,

2521 description="The ID of the constraint.",

2522 ),

2523) -> Response:

2524 """

2525 Get HTML constraint annotation page.

2526

2527 Args:

2528 request (Request): The request context.

2529 project_id (str): The ID of the project.

2530 constraint_id (str): The ID of the constraint.

2531

2532 Returns:

2533 Response: The requested page.

2534 """

2535

2536 # Return HTML constraints page.

2537 try:

2538 return templates.TemplateResponse(

2539 name="constraint_annotation.html",

2540 context={

2541 "request": request,

2542 # Get the project ID.

2543 "project_id": project_id,

2544 # Get the constraints ID.

2545 "constraint_id": constraint_id,

2546 # Get the project metadata (ID, name, creation date).

2547 "metadata": (await get_metadata(project_id=project_id))["metadata"],

2548 # Get the project status (iteration, step name and status, modelization state and conflict).

2549 "status": (await get_status(project_id=project_id))["status"],

2550 # Get the project texts.

2551 "texts": (

2552 await get_texts(

2553 project_id=project_id,

2554 without_deleted_texts=False,

2555 sorted_by=TextsSortOptions.ID,

2556 sorted_reverse=False,

2557 )

2558 )["texts"],

2559 "texts_html_escaped": { # TODO: Escape HTML for javascript

2560 text_id: { # Force HTML escape.

2561 key: html.escape(value) if key in {"text_original", "text", "text_preprocessed"} else value

2562 for key, value in text_value.items()

2563 }

2564 for text_id, text_value in (

2565 await get_texts(

2566 project_id=project_id,

2567 without_deleted_texts=False,

2568 sorted_by=TextsSortOptions.ID,

2569 sorted_reverse=False,

2570 )

2571 )["texts"].items()

2572 },

2573 # Get the project constraints.

2574 "constraints": (

2575 await get_constraints(

2576 project_id=project_id,

2577 without_hidden_constraints=False,

2578 sorted_by=ConstraintsSortOptions.TO_ANNOTATE,

2579 sorted_reverse=False,

2580 )

2581 )["constraints"],

2582 # Get the project clustering result.

2583 "clusters": (await get_constrained_clustering_results(project_id=project_id, iteration_id=None))[

2584 "clustering"

2585 ],

2586 # Get the project modelization inference result.

2587 "modelization": (await get_modelization(project_id=project_id))["modelization"],

2588 },

2589 status_code=status.HTTP_200_OK,

2590 )

2591

2592 # Case of error: Return HTML error page.

2593 except HTTPException as error:

2594 # Return HTML error page.

2595 return templates.TemplateResponse(

2596 name="error.html",

2597 context={

2598 "request": request,

2599 "status_code": error.status_code,

2600 "detail": error.detail,

2601 },

2602 status_code=error.status_code,

2603 )

2604

2605

2606# ==============================================================================

2607# DEFINE ROUTES FOR SETTINGS

2608# ==============================================================================

2609

2610

2611###

2612### ROUTE: Get settings.

2613###

2614@app.get(

2615 "/api/projects/{project_id}/settings",

2616 tags=["Settings"],

2617 status_code=status.HTTP_200_OK,

2618)

2619async def get_settings(

2620 project_id: str = Path(

2621 ...,

2622 description="The ID of the project.",

2623 ),

2624 iteration_id: Optional[int] = Query(

2625 None,

2626 description="The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.",

2627 ),

2628 settings_names: List[ICGUISettings] = Query(

2629 [

2630 ICGUISettings.PREPROCESSING,

2631 ICGUISettings.VECTORIZATION,

2632 ICGUISettings.SAMPLING,

2633 ICGUISettings.CLUSTERING,

2634 ],

2635 description="The list of names of requested settings to return. To select multiple settings kinds, use `CTRL + clic`.",

2636 ),

2637) -> Dict[str, Any]:

2638 """

2639 Get settings.

2640

2641 Args:

2642 project_id (str): The ID of the project.

2643 iteration_id (Optional[int], optional): The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.

2644 settings_names (List[ICGUISettings], optional): The list of names of requested settings to return. Defaults to `[ICGUISettings.PREPROCESSING, ICGUISettings.VECTORIZATION, ICGUISettings.SAMPLING, ICGUISettings.CLUSTERING,]`.

2645

2646 Raises:

2647 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2648 HTTPException: Raises `HTTP_404_NOT_FOUND` if the iteration with id `iteration_id` doesn't exist.

2649

2650 Returns:

2651 Dict[str, Any]: A dictionary that contains settings.

2652 """

2653

2654 # Check project id.

2655 if project_id not in (await get_projects()):

2656 raise HTTPException(

2657 status_code=status.HTTP_404_NOT_FOUND,

2658 detail="The project with id '{project_id_str}' doesn't exist.".format(

2659 project_id_str=str(project_id),

2660 ),

2661 )

2662

2663 # Load settings.

2664 with open(DATA_DIRECTORY / project_id / "settings.json", "r") as settings_fileobject:

2665 project_settings: Dict[str, Dict[str, Any]] = json.load(settings_fileobject)

2666

2667 # Load status file.

2668 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

2669 project_status: Dict[str, Any] = json.load(status_fileobject)

2670

2671 # Get current iteration id if needed.

2672 if iteration_id is None:

2673 iteration_id = project_status["iteration_id"]

2674

2675 # Otherwise check that requested iteration id exist.

2676 if str(iteration_id) not in project_settings.keys():

2677 raise HTTPException(

2678 status_code=status.HTTP_404_NOT_FOUND,

2679 detail="The project with id '{project_id_str}' has no iteration with id '{iteration_id_str}'.".format(

2680 project_id_str=str(project_id),

2681 iteration_id_str=str(iteration_id),

2682 ),

2683 )

2684

2685 # Return the requested settings.

2686 return {

2687 # Get the project ID.

2688 "project_id": project_id,

2689 # Get the iteration ID.

2690 "iteration_id": iteration_id,

2691 # Get the request parameters.

2692 "parameters": {

2693 "settings_names": [settings_name.value for settings_name in settings_names],

2694 },

2695 # Get the settings.

2696 "settings": {

2697 setting_name: settings_value

2698 for setting_name, settings_value in project_settings[str(iteration_id)].items()

2699 if setting_name in settings_names

2700 },

2701 }

2702

2703

2704###

2705### ROUTE: Update settings.

2706###

2707@app.put(

2708 "/api/projects/{project_id}/settings",

2709 tags=["Settings"],

2710 status_code=status.HTTP_201_CREATED,

2711)

2712async def update_settings(

2713 project_id: str = Path(

2714 ...,

2715 description="The ID of the project.",

2716 ),

2717 preprocessing: Optional[PreprocessingSettingsModel] = Body(

2718 None,

2719 description="The settings for data preprocessing. Used during `modelization_update` task. Keep unchanged if empty.",

2720 ),

2721 vectorization: Optional[VectorizationSettingsModel] = Body(

2722 None,

2723 description="The settings for data vectorization. Used during `modelization_update` task. Keep unchanged if empty.",

2724 ),

2725 sampling: Optional[SamplingSettingsModel] = Body(

2726 None,

2727 description="The settings for constraints sampling. Used during `constraints_sampling` task. Keep unchanged if empty.",

2728 ),

2729 clustering: Optional[ClusteringSettingsModel] = Body(

2730 None,

2731 description="The settings for constrained clustering. Used during `constrained_clustering` task. Keep unchanged if empty.",

2732 ),

2733) -> Dict[str, Any]:

2734 """

2735 Update settings.

2736

2737 Args:

2738 project_id (str): The ID of the project.

2739 preprocessing (Optional[PreprocessingSettingsModel], optional): The settings for data preprocessing. Used during `clustering` step. Keep unchanged if empty.. Defaults to None.

2740 vectorization (Optional[VectorizationSettingsModel], optional): The settings for data vectorization. Used during `clustering` step. Keep unchanged if empty.. Defaults to None.

2741 sampling (Optional[SamplingSettingsModel], optional): The settings for constraints sampling. Used during `sampling` step. Keep unchanged if empty.. Defaults to None.

2742 clustering (Optional[ClusteringSettingsModel], optional): The settings for constrained clustering. Used during `clustering` step. Keep unchanged if empty. Defaults to None.

2743

2744 Raises:

2745 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

2746 HTTPException: Raises `HTTP_403_FORBIDDEN` if the status of the project doesn't allow settings modifications.

2747 HTTPException: Raises `HTTP_403_FORBIDDEN` if parameters `preprocessing`, `vectorization`, `sampling` or `clustering` are not expected.

2748 HTTPException: Raises `HTTP_400_BAD_REQUEST` if parameters `preprocessing`, `vectorization`, `sampling` or `clustering` are invalid.

2749

2750 Returns:

2751 Dict[str, Any]: A dictionary that contains the ID of updated settings.

2752 """

2753

2754 # TODO: examples: https://fastapi.tiangolo.com/tutorial/schema-extra-example/#body-with-multiple-examples

2755

2756 # Check project id.

2757 if project_id not in (await get_projects()):

2758 raise HTTPException(

2759 status_code=status.HTTP_404_NOT_FOUND,

2760 detail="The project with id '{project_id_str}' doesn't exist.".format(

2761 project_id_str=str(project_id),

2762 ),

2763 )

2764

2765 # Lock status file in order to check project status for this step.

2766 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

2767 ###

2768 ### Load needed data.

2769 ###

2770

2771 # Load status file.

2772 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject_r:

2773 project_status: Dict[str, Any] = json.load(status_fileobject_r)

2774 iteration_id: int = project_status["iteration_id"]

2775

2776 # Load settings file.

2777 with open(DATA_DIRECTORY / project_id / "settings.json", "r") as settings_fileobject_r:

2778 project_settings: Dict[str, Any] = json.load(settings_fileobject_r)

2779

2780 list_of_updated_settings: List[ICGUISettings] = []

2781

2782 ###

2783 ### Case of preprocessing settings.

2784 ###

2785 if preprocessing is not None:

2786 list_of_updated_settings.append(ICGUISettings.PREPROCESSING)

2787

2788 # Check project status for preprocessing.

2789 if (

2790 project_status["state"] != ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION # noqa: WPS514

2791 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION

2792 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2793 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2794 ):

2795 raise HTTPException(

2796 status_code=status.HTTP_403_FORBIDDEN,

2797 detail="The 'preprocessing' settings of project with id '{project_id_str}' cant't be modified during this state (state='{state_str}'). No changes have been taken into account.".format(

2798 project_id_str=str(project_id),

2799 state_str=str(project_status["state"]),

2800 ),

2801 )

2802

2803 # Update status by forcing "outdated" status.

2804 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

2805 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2806 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

2807 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2808 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

2809 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2810 #### elif project_status["state"] == ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION:

2811 #### project_status["state"] = ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION

2812

2813 # Update the default settings with the parameters in the request body.

2814 for key_prep, value_prep in preprocessing.to_dict().items():

2815 project_settings[str(iteration_id)]["preprocessing"][key_prep] = value_prep

2816

2817 ###

2818 ### Case of vectorization settings.

2819 ###

2820 if vectorization is not None:

2821 list_of_updated_settings.append(ICGUISettings.VECTORIZATION)

2822

2823 # Check project status for vectorization.

2824 if (

2825 project_status["state"] != ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION # noqa: WPS514

2826 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION

2827 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2828 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2829 ):

2830 raise HTTPException(

2831 status_code=status.HTTP_403_FORBIDDEN,

2832 detail="The 'vectorization' settings of project with id '{project_id_str}' cant't be modified during this state (state='{state_str}'). No changes have been taken into account.".format(

2833 project_id_str=str(project_id),

2834 state_str=str(project_status["state"]),

2835 ),

2836 )

2837

2838 # Update status by forcing "outdated" status.

2839 if project_status["state"] == ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION:

2840 project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2841 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

2842 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2843 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

2844 #### project_status["state"] = ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2845 #### elif project_status["state"] == ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION:

2846 #### project_status["state"] = ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION

2847

2848 # Update the default settings with the parameters in the request body.

2849 for key_vect, value_vect in vectorization.to_dict().items():

2850 project_settings[str(iteration_id)]["vectorization"][key_vect] = value_vect

2851

2852 ###

2853 ### Case of sampling settings.

2854 ###

2855 if sampling is not None:

2856 list_of_updated_settings.append(ICGUISettings.SAMPLING)

2857

2858 # Check project status for sampling.

2859 if project_status["state"] != ICGUIStates.SAMPLING_TODO:

2860 raise HTTPException(

2861 status_code=status.HTTP_403_FORBIDDEN,

2862 detail="The 'sampling' settings of project with id '{project_id_str}' cant't be modified during this state (state='{state_str}'). No changes have been taken into account.".format(

2863 project_id_str=str(project_id),

2864 state_str=str(project_status["state"]),

2865 ),

2866 )

2867

2868 # Update the default settings with the parameters in the request body.

2869 for key_sampl, value_sampl in sampling.to_dict().items():

2870 project_settings[str(iteration_id)]["sampling"][key_sampl] = value_sampl

2871

2872 ###

2873 ### Case of clustering settings.

2874 ###

2875 if clustering is not None:

2876 list_of_updated_settings.append(ICGUISettings.CLUSTERING)

2877

2878 # Check project status for clustering.

2879 if (

2880 project_status["state"] != ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION # noqa: WPS514

2881 and project_status["state"] != ICGUIStates.INITIALIZATION_WITH_PENDING_MODELIZATION

2882 and project_status["state"] != ICGUIStates.INITIALIZATION_WITH_WORKING_MODELIZATION

2883 and project_status["state"] != ICGUIStates.SAMPLING_TODO

2884 and project_status["state"] != ICGUIStates.SAMPLING_PENDING

2885 and project_status["state"] != ICGUIStates.SAMPLING_WORKING

2886 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION

2887 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

2888 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITHOUT_CONFLICTS

2889 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITHOUT_CONFLICTS

2890 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

2891 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITH_CONFLICTS

2892 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITH_CONFLICTS

2893 and project_status["state"] != ICGUIStates.CLUSTERING_TODO

2894 ):

2895 raise HTTPException(

2896 status_code=status.HTTP_403_FORBIDDEN,

2897 detail="The 'clustering' settings of project with id '{project_id_str}' cant't be modified during this state (state='{state_str}'). No changes have been taken into account.".format(

2898 project_id_str=str(project_id),

2899 state_str=str(project_status["state"]),

2900 ),

2901 )

2902

2903 # Update the default settings with the parameters in the request body.

2904 for key_clus, value_clus in clustering.to_dict().items():

2905 project_settings[str(iteration_id)]["clustering"][key_clus] = value_clus

2906

2907 ###

2908 ### Store updated data.

2909 ###

2910

2911 # Store updated status in file.

2912 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

2913 json.dump(project_status, status_fileobject_w, indent=4)

2914

2915 # Store updated settings in file.

2916 with open(DATA_DIRECTORY / project_id / "settings.json", "w") as settings_fileobject_w:

2917 json.dump(project_settings, settings_fileobject_w, indent=4)

2918

2919 ###

2920 ### Return statement.

2921 ###

2922 return {

2923 "project_id": project_id,

2924 "detail": "The project with id '{project_id_str}' has updated the following settings: {settings_str}.".format(

2925 project_id_str=str(project_id),

2926 settings_str=", ".join(list_of_updated_settings),

2927 ),

2928 }

2929

2930

2931###

2932### ROUTE: Get HTML settings page.

2933###

2934@app.get(

2935 "/gui/projects/{project_id}/settings",

2936 tags=["Settings"],

2937 response_class=Response,

2938 status_code=status.HTTP_200_OK,

2939)

2940async def get_html_settings_page(

2941 request: Request,

2942 project_id: str = Path(

2943 ...,

2944 description="The ID of the project.",

2945 ),

2946 iteration_id: Optional[int] = Query(

2947 None,

2948 description="The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.",

2949 ),

2950 settings_names: List[ICGUISettings] = Query(

2951 [

2952 ICGUISettings.PREPROCESSING,

2953 ICGUISettings.VECTORIZATION,

2954 ICGUISettings.SAMPLING,

2955 ICGUISettings.CLUSTERING,

2956 ],

2957 description="The list of names of requested settings to return. To select multiple settings kinds, use `CTRL + clic`.",

2958 ),

2959) -> Response:

2960 """

2961 Get HTML settings page.

2962

2963 Args:

2964 request (Request): The request context.

2965 project_id (str): The ID of the project.

2966 iteration_id (Optional[int], optional): The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.

2967 settings_names (List[ICGUISettings], optional): The list of names of requested settings to return. Defaults to `[ICGUISettings.PREPROCESSING, ICGUISettings.VECTORIZATION, ICGUISettings.SAMPLING, ICGUISettings.CLUSTERING,]`.

2968

2969 Returns:

2970 Response: The requested page.

2971 """

2972

2973 # Return HTML project home page.

2974 try: # noqa: WPS229 (too long try body)

2975 project_status: Dict[str, Any] = (await get_status(project_id=project_id))["status"]

2976 if iteration_id is None:

2977 iteration_id = project_status["iteration_id"]

2978

2979 return templates.TemplateResponse(

2980 name="settings.html",

2981 context={

2982 "request": request,

2983 # Get the project ID.

2984 "project_id": project_id,

2985 # Get the iteration ID.

2986 "iteration_id": iteration_id,

2987 # Get the request parameters.

2988 "parameters": {

2989 "settings_names": [settings_name.value for settings_name in settings_names],

2990 },

2991 # Get the project metadata (ID, name, creation date).

2992 "metadata": (await get_metadata(project_id=project_id))["metadata"],

2993 # Get the project status (iteration, step name and status, modelization state and conflict).

2994 "status": project_status,

2995 # Get the project settings (preprocessing, vectorization, sampling, clustering).

2996 "settings": (

2997 await get_settings(project_id=project_id, iteration_id=iteration_id, settings_names=settings_names)

2998 )["settings"],

2999 # Get navigation information.

3000 "navigation": {

3001 "previous": None if (iteration_id == 0) else iteration_id - 1,

3002 "next": None if (iteration_id == project_status["iteration_id"]) else (iteration_id + 1),

3003 },

3004 },

3005 status_code=status.HTTP_200_OK,

3006 )

3007

3008 # Case of error: Return HTML error page.

3009 except HTTPException as error:

3010 # Return HTML error page.

3011 return templates.TemplateResponse(

3012 name="error.html",

3013 context={

3014 "request": request,

3015 "status_code": error.status_code,

3016 "detail": error.detail,

3017 },

3018 status_code=error.status_code,

3019 )

3020

3021

3022# ==============================================================================

3023# DEFINE ROUTES FOR MODELIZATION

3024# ==============================================================================

3025

3026

3027###

3028### ROUTE: Get modelization inference.

3029###

3030@app.get(

3031 "/api/projects/{project_id}/modelization",

3032 tags=["Data modelization"],

3033 status_code=status.HTTP_200_OK,

3034)

3035async def get_modelization(

3036 project_id: str = Path(

3037 ...,

3038 description="The ID of the project.",

3039 ),

3040) -> Dict[str, Any]:

3041 """

3042 Get modelization inference.

3043

3044 Args:

3045 project_id (str, optional): The ID of the project.

3046

3047 Raises:

3048 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3049

3050 Returns:

3051 Dict[str, Any]: A dictionary that contains modelization inference result.

3052 """

3053

3054 # Check project id.

3055 if project_id not in (await get_projects()):

3056 raise HTTPException(

3057 status_code=status.HTTP_404_NOT_FOUND,

3058 detail="The project with id '{project_id_str}' doesn't exist.".format(

3059 project_id_str=str(project_id),

3060 ),

3061 )

3062

3063 # Load the modelization inference results.

3064 with open(DATA_DIRECTORY / project_id / "modelization.json", "r") as modelization_fileobject:

3065 # Return the project modelization inference.

3066 return {

3067 "project_id": project_id,

3068 "modelization": json.load(modelization_fileobject),

3069 }

3070

3071

3072###

3073### ROUTE: Get 2D and 3D vectors.

3074###

3075@app.get(

3076 "/api/projects/{project_id}/vectors",

3077 tags=["Data modelization"],

3078 status_code=status.HTTP_200_OK,

3079)

3080async def get_vectors(

3081 project_id: str = Path(

3082 ...,

3083 description="The ID of the project.",

3084 ),

3085) -> Dict[str, Any]:

3086 """

3087 Get 2D and 3D vectors.

3088

3089 Args:

3090 project_id (str, optional): The ID of the project.

3091

3092 Raises:

3093 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3094 HTTPException: Raises `HTTP_404_NOT_FOUND` if the iteration with id `iteration_id` doesn't exist.

3095 HTTPException: Raises `HTTP_403_FORBIDDEN` if the status of the project hasn't completed its clustering step.

3096

3097 Returns:

3098 Dict[str, Any]: A dictionary that contains clustering result.

3099 """

3100

3101 # Check project id.

3102 if project_id not in (await get_projects()):

3103 raise HTTPException(

3104 status_code=status.HTTP_404_NOT_FOUND,

3105 detail="The project with id '{project_id_str}' doesn't exist.".format(

3106 project_id_str=str(project_id),

3107 ),

3108 )

3109

3110 # Load status file.

3111 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3112 project_status: Dict[str, Any] = json.load(status_fileobject)

3113

3114 # Check project status.

3115 if (

3116 project_status["state"] != ICGUIStates.SAMPLING_TODO # noqa: WPS514

3117 and project_status["state"] != ICGUIStates.SAMPLING_PENDING

3118 and project_status["state"] != ICGUIStates.SAMPLING_WORKING

3119 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_UPTODATE_MODELIZATION

3120 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

3121 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITHOUT_CONFLICTS

3122 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITHOUT_CONFLICTS

3123 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

3124 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITH_CONFLICTS

3125 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_WORKING_MODELIZATION_WITH_CONFLICTS

3126 and project_status["state"] != ICGUIStates.CLUSTERING_TODO

3127 and project_status["state"] != ICGUIStates.CLUSTERING_PENDING

3128 and project_status["state"] != ICGUIStates.CLUSTERING_WORKING

3129 and project_status["state"] != ICGUIStates.ITERATION_END

3130 ):

3131 raise HTTPException(

3132 status_code=status.HTTP_403_FORBIDDEN,

3133 detail="The project with id '{project_id_str}' hasn't completed its modelization update step.".format(

3134 project_id_str=str(project_id),

3135 ),

3136 )

3137

3138 # Load the 2D vectors.

3139 with open(DATA_DIRECTORY / project_id / "vectors_2D.json", "r") as vectors_2D_fileobject:

3140 vectors_2D: Dict[str, Dict[str, float]] = json.load(vectors_2D_fileobject) # noqa: S301 # Usage of Pickle

3141

3142 # Load the 3D vectors.

3143 with open(DATA_DIRECTORY / project_id / "vectors_3D.json", "r") as vectors_3D_fileobject:

3144 vectors_3D: Dict[str, Dict[str, float]] = json.load(vectors_3D_fileobject) # noqa: S301 # Usage of Pickle

3145

3146 # Return the project vectors.

3147 return {

3148 "project_id": project_id,

3149 "vectors_2d": vectors_2D,

3150 "vectors_3d": vectors_3D,

3151 }

3152

3153

3154###

3155### ROUTE: Prepare modelization update task.

3156###

3157@app.post(

3158 "/api/projects/{project_id}/modelization",

3159 tags=["Data modelization"],

3160 status_code=status.HTTP_202_ACCEPTED,

3161)

3162async def prepare_modelization_update_task(

3163 background_tasks: BackgroundTasks,

3164 project_id: str = Path(

3165 ...,

3166 description="The ID of the project.",

3167 ),

3168) -> Dict[str, Any]:

3169 """

3170 Prepare modelization update task.

3171

3172 Args:

3173 background_tasks (BackgroundTasks): A background task to run after the return statement.

3174 project_id (str): The ID of the project.

3175

3176 Raises:

3177 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3178 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow the preparation of modelization update task.

3179

3180 Returns:

3181 Dict[str, Any]: A dictionary that contains the confirmation of the preparation of modelization update task.

3182 """

3183

3184 # Check project id.

3185 if project_id not in (await get_projects()):

3186 raise HTTPException(

3187 status_code=status.HTTP_404_NOT_FOUND,

3188 detail="The project with id '{project_id_str}' doesn't exist.".format(

3189 project_id_str=str(project_id),

3190 ),

3191 )

3192

3193 # Lock status file in order to check project status for this step.

3194 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

3195 ###

3196 ### Load needed data.

3197 ###

3198

3199 # Load status file.

3200 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3201 project_status: Dict[str, Any] = json.load(status_fileobject)

3202

3203 ###

3204 ### Check parameters.

3205 ###

3206

3207 # Check status.

3208 if (

3209 project_status["state"] != ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION # noqa: WPS514

3210 and project_status["state"] != ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION

3211 and project_status["state"] != ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITHOUT_MODELIZATION

3212 and project_status["state"] != ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITHOUT_MODELIZATION

3213 and project_status["state"] != ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION

3214 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS

3215 and project_status["state"] != ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS

3216 ):

3217 raise HTTPException(

3218 status_code=status.HTTP_403_FORBIDDEN,

3219 detail="The project with id '{project_id_str}' doesn't allow the preparation of modelization update task during this state (state='{state_str}').".format(

3220 project_id_str=str(project_id),

3221 state_str=str(project_status["state"]),

3222 ),

3223 )

3224

3225 ###

3226 ### Update data.

3227 ###

3228

3229 # Update status by forcing "pending" status.

3230 if project_status["state"] == ICGUIStates.INITIALIZATION_WITHOUT_MODELIZATION:

3231 project_status["state"] = ICGUIStates.INITIALIZATION_WITH_PENDING_MODELIZATION

3232 elif project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION:

3233 project_status["state"] = ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_PENDING_MODELIZATION

3234 elif project_status["state"] == ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITHOUT_MODELIZATION:

3235 project_status["state"] = ICGUIStates.IMPORT_AT_ANNOTATION_STEP_WITH_PENDING_MODELIZATION

3236 elif project_status["state"] == ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITHOUT_MODELIZATION:

3237 project_status["state"] = ICGUIStates.IMPORT_AT_CLUSTERING_STEP_WITH_PENDING_MODELIZATION

3238 elif project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION:

3239 project_status["state"] = ICGUIStates.IMPORT_AT_ITERATION_END_WITH_PENDING_MODELIZATION

3240 elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITHOUT_CONFLICTS:

3241 project_status["state"] = ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITHOUT_CONFLICTS

3242 #### elif project_status["state"] == ICGUIStates.ANNOTATION_WITH_OUTDATED_MODELIZATION_WITH_CONFLICTS:

3243 else:

3244 project_status["state"] = ICGUIStates.ANNOTATION_WITH_PENDING_MODELIZATION_WITH_CONFLICTS

3245

3246 # Prepare status by initializing "task" status.

3247 project_status["task"] = {

3248 "progression": 1,

3249 "detail": "Waiting for background task allocation...",

3250 }

3251

3252 ###

3253 ### Store updated data.

3254 ###

3255

3256 # Store updated status in file.

3257 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

3258 json.dump(project_status, status_fileobject_w, indent=4)

3259

3260 ###

3261 ### Launch backgroundtask.

3262 ###

3263

3264 # Add the background task.

3265 background_tasks.add_task(

3266 func=backgroundtasks.run_modelization_update_task,

3267 project_id=project_id,

3268 )

3269

3270 # Return statement.

3271 return { # pragma: no cover (need radis and worder)

3272 "project_id": project_id,

3273 "detail": "In project with id '{project_id_str}', the modelization update task has been requested and is waiting for a background task.".format(

3274 project_id_str=str(project_id),

3275 ),

3276 }

3277

3278

3279# ==============================================================================

3280# DEFINE ROUTES FOR CONSTRAINTS SAMPLING

3281# ==============================================================================

3282

3283

3284###

3285### ROUTE: Get constraints sampling results.

3286###

3287@app.get(

3288 "/api/projects/{project_id}/sampling",

3289 tags=["Constraints sampling"],

3290 status_code=status.HTTP_200_OK,

3291)

3292async def get_constraints_sampling_results(

3293 project_id: str = Path(

3294 ...,

3295 description="The ID of the project.",

3296 ),

3297 iteration_id: Optional[int] = Query(

3298 None,

3299 description="The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.",

3300 ),

3301) -> Dict[str, Any]:

3302 """

3303 Get constraints sampling results.

3304

3305 Args:

3306 project_id (str, optional): The ID of the project.

3307 iteration_id (Optional[int], optional): The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.

3308

3309 Raises:

3310 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3311 HTTPException: Raises `HTTP_404_NOT_FOUND` if the iteration with id `iteration_id` doesn't exist.

3312 HTTPException: Raises `HTTP_403_FORBIDDEN` if the status of the project hasn't completed its sampling step.

3313

3314 Returns:

3315 Dict[str, Any]: A dictionary that contains sampling result.

3316 """

3317

3318 # Check project id.

3319 if project_id not in (await get_projects()):

3320 raise HTTPException(

3321 status_code=status.HTTP_404_NOT_FOUND,

3322 detail="The project with id '{project_id_str}' doesn't exist.".format(

3323 project_id_str=str(project_id),

3324 ),

3325 )

3326

3327 # Load settings.

3328 with open(DATA_DIRECTORY / project_id / "settings.json", "r") as settings_fileobject:

3329 project_settings: Dict[str, Dict[str, Any]] = json.load(settings_fileobject)

3330

3331 # Load status file.

3332 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3333 project_status: Dict[str, Any] = json.load(status_fileobject)

3334

3335 # Get current iteration id if needed.

3336 if iteration_id is None:

3337 if project_status["iteration_id"] == 0:

3338 iteration_id = 0

3339 elif (

3340 project_status["state"] == ICGUIStates.SAMPLING_TODO # noqa: WPS514

3341 or project_status["state"] == ICGUIStates.SAMPLING_PENDING

3342 or project_status["state"] == ICGUIStates.SAMPLING_WORKING

3343 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION

3344 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_PENDING_MODELIZATION

3345 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_WORKING_MODELIZATION

3346 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_ERRORS

3347 ):

3348 iteration_id = project_status["iteration_id"] - 1

3349 else:

3350 iteration_id = project_status["iteration_id"]

3351

3352 # Case of iteration `0`.

3353 if iteration_id == 0:

3354 raise HTTPException(

3355 status_code=status.HTTP_403_FORBIDDEN,

3356 detail="The iteration `0` has no sampling step.",

3357 )

3358

3359 # Check project status.

3360 if iteration_id == project_status["iteration_id"] and (

3361 project_status["state"] == ICGUIStates.SAMPLING_TODO # noqa: WPS514

3362 or project_status["state"] == ICGUIStates.SAMPLING_PENDING

3363 or project_status["state"] == ICGUIStates.SAMPLING_WORKING

3364 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITHOUT_MODELIZATION

3365 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_PENDING_MODELIZATION

3366 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_WORKING_MODELIZATION

3367 or project_status["state"] == ICGUIStates.IMPORT_AT_SAMPLING_STEP_WITH_ERRORS

3368 ):

3369 raise HTTPException(

3370 status_code=status.HTTP_403_FORBIDDEN,

3371 detail="The project with id '{project_id_str}' hasn't completed its sampling step on iteration '{iteration_id_str}'.".format(

3372 project_id_str=str(project_id),

3373 iteration_id_str=str(iteration_id),

3374 ),

3375 )

3376

3377 # Otherwise check that requested iteration id exist.

3378 if str(iteration_id) not in project_settings.keys():

3379 raise HTTPException(

3380 status_code=status.HTTP_404_NOT_FOUND,

3381 detail="The project with id '{project_id_str}' has no iteration with id '{iteration_id_str}'.".format(

3382 project_id_str=str(project_id),

3383 iteration_id_str=str(iteration_id),

3384 ),

3385 )

3386

3387 # Load the sampling results.

3388 with open(DATA_DIRECTORY / project_id / "sampling.json", "r") as sampling_fileobject:

3389 # Return the project sampling.

3390 return {

3391 "project_id": project_id,

3392 "iteration_id": iteration_id,

3393 "sampling": json.load(sampling_fileobject)[str(iteration_id)],

3394 }

3395

3396

3397###

3398### ROUTE: Prepare constraints sampling task.

3399###

3400@app.post(

3401 "/api/projects/{project_id}/sampling",

3402 tags=["Constraints sampling"],

3403 status_code=status.HTTP_202_ACCEPTED,

3404)

3405async def prepare_constraints_sampling_task(

3406 background_tasks: BackgroundTasks,

3407 project_id: str = Path(

3408 ...,

3409 description="The ID of the project.",

3410 ),

3411) -> Dict[str, Any]:

3412 """

3413 Prepare constraints sampling task.

3414

3415 Args:

3416 background_tasks (BackgroundTasks): A background task to run after the return statement.

3417 project_id (str): The ID of the project.

3418

3419 Raises:

3420 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3421 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow the preparation of constraints sampling task.

3422

3423 Returns:

3424 Dict[str, Any]: A dictionary that contains the confirmation of the preparation of constraints sampling task.

3425 """

3426

3427 # Check project id.

3428 if project_id not in (await get_projects()):

3429 raise HTTPException(

3430 status_code=status.HTTP_404_NOT_FOUND,

3431 detail="The project with id '{project_id_str}' doesn't exist.".format(

3432 project_id_str=str(project_id),

3433 ),

3434 )

3435

3436 # Lock status file in order to check project status for this step.

3437 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

3438 # Load status file.

3439 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3440 project_status: Dict[str, Any] = json.load(status_fileobject)

3441

3442 # Check status.

3443 if project_status["state"] != ICGUIStates.SAMPLING_TODO:

3444 raise HTTPException(

3445 status_code=status.HTTP_403_FORBIDDEN,

3446 detail="The project with id '{project_id_str}' doesn't allow the preparation of constraints sampling task during this state (state='{state_str}').".format(

3447 project_id_str=str(project_id),

3448 state_str=str(project_status["state"]),

3449 ),

3450 )

3451

3452 ###

3453 ### Update data.

3454 ###

3455

3456 # Update status by forcing "pending" status.

3457 project_status["state"] = ICGUIStates.SAMPLING_PENDING

3458

3459 # Prepare status by initializing "task" status.

3460 project_status["task"] = {

3461 "progression": 1,

3462 "detail": "Waiting for background task allocation...",

3463 }

3464

3465 ###

3466 ### Store updated data.

3467 ###

3468

3469 # Store updated status in file.

3470 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

3471 json.dump(project_status, status_fileobject_w, indent=4)

3472

3473 ###

3474 ### Launch backgroundtask.

3475 ###

3476

3477 # Add the background task.

3478 background_tasks.add_task(

3479 func=backgroundtasks.run_constraints_sampling_task,

3480 project_id=project_id,

3481 )

3482

3483 # Return statement.

3484 return { # pragma: no cover (need radis and worder)

3485 "project_id": project_id,

3486 "detail": "In project with id '{project_id_str}', the constraints sampling task has been requested and is waiting for a background task.".format(

3487 project_id_str=str(project_id),

3488 ),

3489 }

3490

3491

3492# ==============================================================================

3493# DEFINE ROUTES FOR CONSTRAINED CLUSTERING

3494# ==============================================================================

3495

3496

3497###

3498### ROUTE: Get constrained clustering results.

3499###

3500@app.get(

3501 "/api/projects/{project_id}/clustering",

3502 tags=["Constrained clustering"],

3503 status_code=status.HTTP_200_OK,

3504)

3505async def get_constrained_clustering_results(

3506 project_id: str = Path(

3507 ...,

3508 description="The ID of the project.",

3509 ),

3510 iteration_id: Optional[int] = Query(

3511 None,

3512 description="The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.",

3513 ),

3514) -> Dict[str, Any]:

3515 """

3516 Get constrained clustering results.

3517

3518 Args:

3519 project_id (str, optional): The ID of the project.

3520 iteration_id (Optional[int], optional): The ID of project iteration. If `None`, get the current iteration. Defaults to `None`.

3521

3522 Raises:

3523 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3524 HTTPException: Raises `HTTP_404_NOT_FOUND` if the iteration with id `iteration_id` doesn't exist.

3525 HTTPException: Raises `HTTP_403_FORBIDDEN` if the status of the project hasn't completed its clustering step.

3526

3527 Returns:

3528 Dict[str, Any]: A dictionary that contains clustering result.

3529 """

3530

3531 # Check project id.

3532 if project_id not in (await get_projects()):

3533 raise HTTPException(

3534 status_code=status.HTTP_404_NOT_FOUND,

3535 detail="The project with id '{project_id_str}' doesn't exist.".format(

3536 project_id_str=str(project_id),

3537 ),

3538 )

3539

3540 # Load status file.

3541 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3542 project_status: Dict[str, Any] = json.load(status_fileobject)

3543

3544 # Load clustering.

3545 with open(DATA_DIRECTORY / project_id / "clustering.json", "r") as clustering_fileobject:

3546 project_clustering: Dict[str, Dict[str, Any]] = json.load(clustering_fileobject)

3547

3548 # Set iteration id if needed.

3549 if iteration_id is None:

3550 if (

3551 project_status["iteration_id"] == 0

3552 or project_status["state"] == ICGUIStates.ITERATION_END

3553 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION

3554 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_PENDING_MODELIZATION

3555 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_WORKING_MODELIZATION

3556 or project_status["state"] == ICGUIStates.IMPORT_AT_ITERATION_END_WITH_ERRORS

3557 ):

3558 iteration_id = project_status["iteration_id"]

3559 else:

3560 iteration_id = project_status["iteration_id"] - 1

3561

3562 # Check project status.

3563 if (

3564 iteration_id == project_status["iteration_id"]

3565 and project_status["state"] != ICGUIStates.ITERATION_END

3566 and project_status["state"] != ICGUIStates.IMPORT_AT_ITERATION_END_WITHOUT_MODELIZATION

3567 and project_status["state"] != ICGUIStates.IMPORT_AT_ITERATION_END_WITH_PENDING_MODELIZATION

3568 and project_status["state"] != ICGUIStates.IMPORT_AT_ITERATION_END_WITH_WORKING_MODELIZATION

3569 and project_status["state"] != ICGUIStates.IMPORT_AT_ITERATION_END_WITH_ERRORS

3570 ):

3571 raise HTTPException(

3572 status_code=status.HTTP_403_FORBIDDEN,

3573 detail="The project with id '{project_id_str}' hasn't completed its clustering step on iteration '{iteration_id_str}'.".format(

3574 project_id_str=str(project_id),

3575 iteration_id_str=str(iteration_id),

3576 ),

3577 )

3578

3579 # Otherwise check that requested iteration id exist.

3580 if str(iteration_id) not in project_clustering.keys():

3581 raise HTTPException(

3582 status_code=status.HTTP_404_NOT_FOUND,

3583 detail="The project with id '{project_id_str}' has no iteration with id '{iteration_id_str}'.".format(

3584 project_id_str=str(project_id),

3585 iteration_id_str=str(iteration_id),

3586 ),

3587 )

3588

3589 # Return the project clustering.

3590 return {

3591 "project_id": project_id,

3592 "iteration_id": iteration_id,

3593 "clustering": project_clustering[str(iteration_id)],

3594 }

3595

3596

3597###

3598### ROUTE: Prepare constrained clustering task.

3599###

3600@app.post(

3601 "/api/projects/{project_id}/clustering",

3602 tags=["Constrained clustering"],

3603 status_code=status.HTTP_202_ACCEPTED,

3604)

3605async def prepare_constrained_clustering_task(

3606 background_tasks: BackgroundTasks,

3607 project_id: str = Path(

3608 ...,

3609 description="The ID of the project.",

3610 ),

3611) -> Dict[str, Any]:

3612 """

3613 Prepare constrained clustering task.

3614

3615 Args:

3616 background_tasks (BackgroundTasks): A background task to run after the return statement.

3617 project_id (str): The ID of the project.

3618

3619 Raises:

3620 HTTPException: Raises `HTTP_404_NOT_FOUND` if the project with id `project_id` doesn't exist.

3621 HTTPException: Raises `HTTP_403_FORBIDDEN` if the current status of the project doesn't allow the preparation of constrained clustering task.

3622 HTTPException: Raises `HTTP_504_GATEWAY_TIMEOUT` if the task can't be prepared.

3623

3624 Returns:

3625 Dict[str, Any]: A dictionary that contains the confirmation of the preparation of constrained clustering task.

3626 """

3627

3628 # Check project id.

3629 if project_id not in (await get_projects()):

3630 raise HTTPException(

3631 status_code=status.HTTP_404_NOT_FOUND,

3632 detail="The project with id '{project_id_str}' doesn't exist.".format(

3633 project_id_str=str(project_id),

3634 ),

3635 )

3636

3637 # Lock status file in order to check project status for this step.

3638 with FileLock(str(DATA_DIRECTORY / project_id / "status.json.lock")):

3639 ###

3640 ### Load needed data.

3641 ###

3642

3643 # Load status file.

3644 with open(DATA_DIRECTORY / project_id / "status.json", "r") as status_fileobject:

3645 project_status: Dict[str, Any] = json.load(status_fileobject)

3646

3647 ###

3648 ### Check parameters.

3649 ###

3650

3651 # Check status.

3652 if project_status["state"] != ICGUIStates.CLUSTERING_TODO:

3653 raise HTTPException(

3654 status_code=status.HTTP_403_FORBIDDEN,

3655 detail="The project with id '{project_id_str}' doesn't allow the preparation of constrained clustering task during this state (state='{state_str}').".format(

3656 project_id_str=str(project_id),

3657 state_str=str(project_status["state"]),

3658 ),

3659 )

3660

3661 ###

3662 ### Update data.

3663 ###

3664

3665 # Update status by forcing "pending" status.

3666 project_status["state"] = ICGUIStates.CLUSTERING_PENDING

3667

3668 # Prepare status by initializing "task" status.

3669 project_status["task"] = {

3670 "progression": 1,

3671 "detail": "Waiting for background task allocation...",

3672 }

3673

3674 ###

3675 ### Store updated data.

3676 ###

3677

3678 # Store updated status in file.

3679 with open(DATA_DIRECTORY / project_id / "status.json", "w") as status_fileobject_w:

3680 json.dump(project_status, status_fileobject_w, indent=4)

3681

3682 ###

3683 ### Launch backgroundtask.

3684 ###

3685

3686 # Add the background task.

3687 background_tasks.add_task(

3688 func=backgroundtasks.run_constrained_clustering_task,

3689 project_id=project_id,

3690 )

3691

3692 # Return statement.

3693 return { # pragma: no cover (need radis and worder)

3694 "project_id": project_id,

3695 "detail": "In project with id '{project_id_str}', the constrained clustering task has been requested and is waiting for a background task.".format(

3696 project_id_str=str(project_id),

3697 ),

3698 }

Coverage for src\cognitivefactory\interactive_clustering_gui\app.py: 100.00%

635 statements