diff --git a/.gitattributes b/.gitattributes index 1c17d4cb..9fb06228 100644 --- a/.gitattributes +++ b/.gitattributes @@ -5,3 +5,4 @@ *.onnx filter=lfs diff=lfs merge=lfs -text *.JPG filter=lfs diff=lfs merge=lfs -text *.png filter=lfs diff=lfs merge=lfs -text +*.db filter=lfs diff=lfs merge=lfs -text diff --git a/.github/workflows/python-build.yml b/.github/workflows/python-build.yml index 870b3aac..c83cc36e 100644 --- a/.github/workflows/python-build.yml +++ b/.github/workflows/python-build.yml @@ -9,6 +9,8 @@ jobs: steps: - uses: actions/checkout@v5 + with: + lfs: true - name: Set up Python uses: actions/setup-python@v6 diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index 8a4849e1..62791933 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -29,6 +29,8 @@ jobs: steps: - uses: actions/checkout@v5 + with: + lfs: true - name: Download wheel artifact uses: actions/download-artifact@v8 diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index 14547a7b..e1ee3e2d 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -16,6 +16,8 @@ jobs: steps: - uses: actions/checkout@v5 + with: + lfs: true - name: Set up Python uses: actions/setup-python@v6 diff --git a/python_sdk/pyproject.toml b/python_sdk/pyproject.toml index 92ec6333..68c382d4 100644 --- a/python_sdk/pyproject.toml +++ b/python_sdk/pyproject.toml @@ -15,9 +15,9 @@ authors = [ description = "This package contains a SDK for Reality Modeling, Reality Analysis and Reality Conversion iTwin APIs as well as Reality Management API utils. It provides classes, functions and examples to upload local data to ContextShare, run jobs and download the results." readme = "README.md" license = {text = "MIT"} -requires-python = ">=3.8" +requires-python = ">=3.10" classifiers = [ - "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.10", "License :: OSI Approved :: MIT License", "Operating System :: Microsoft :: Windows", ] @@ -48,3 +48,7 @@ dev = [ "Reality Modeling Api reference" = "https://developer.bentley.com/apis/contextcapture/" "Reality Conversion Api reference" = "https://developer.bentley.com/apis/realityconversion/" "Reality Management Api reference" = "https://developer.bentley.com/apis/reality-management/" + +[tool.coverage.run] +source = ["reality_capture", "docs"] +omit = ["tests/*", "docs/conf.py"] \ No newline at end of file diff --git a/python_sdk/src/reality_capture/common/__init__.py b/python_sdk/src/reality_capture/common/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python_sdk/src/reality_capture/common/job.py b/python_sdk/src/reality_capture/common/job.py new file mode 100644 index 00000000..c7623454 --- /dev/null +++ b/python_sdk/src/reality_capture/common/job.py @@ -0,0 +1,52 @@ +from pydantic import BaseModel, Field +from datetime import datetime +from enum import Enum +from typing import Optional + + +class JobType(Enum): + CALIBRATION = "Calibration" + CHANGE_DETECTION = "ChangeDetection" + CONSTRAINTS = "Constraints" + EVAL_O2D = "EvalO2D" + EVAL_O3D = "EvalO3D" + EVAL_S2D = "EvalS2D" + EVAL_S3D = "EvalS3D" + EVAL_SORTHO = "EvalSOrtho" + FILL_IMAGE_PROPERTIES = "FillImageProperties" + GAUSSIAN_SPLATS = "GaussianSplats" + IMPORT_POINT_CLOUD = "ImportPointCloud" + OBJECTS_2D = "Objects2D" + PRODUCTION = "Production" + RECONSTRUCTION = "Reconstruction" + SEGMENTATION_2D = "Segmentation2D" + SEGMENTATION_3D = "Segmentation3D" + SEGMENTATION_ORTHOPHOTO = "SegmentationOrthophoto" + TILING = "Tiling" + TOUCH_UP_IMPORT = "TouchUpImport" + TOUCH_UP_EXPORT = "TouchUpExport" + WATER_CONSTRAINTS = "WaterConstraints" + CLEARANCE_CALCULATION = "ClearanceCalculation" + # POINT_CLOUD_CONVERSION = "PointCloudConversion" + + +class JobState(Enum): + QUEUED = "Queued" + ACTIVE = "Active" + SUCCESS = "Success" + FAILED = "Failed" + TERMINATING_ON_CANCEL = "TerminatingOnCancel" + TERMINATING_ON_FAILURE = "TerminatingOnFailure" + CANCELLED = "Cancelled" + + +class BaseExecution(BaseModel): + created_date_time: datetime = Field(description="Creation date time for the job.", alias="createdDateTime") + started_date_time: Optional[datetime] = Field(None, description="Start date time for the job.", + alias="startedDateTime") + ended_date_time: Optional[datetime] = Field(None, description="End date time for the job.", alias="endedDateTime") + + +class BaseProgress(BaseModel): + state: JobState = Field(description="State of the job.") + percentage: float = Field(ge=0, le=100, description="Progress of the job.") diff --git a/python_sdk/src/reality_capture/on_premise/__init__.py b/python_sdk/src/reality_capture/on_premise/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python_sdk/src/reality_capture/on_premise/_generic_manager.py b/python_sdk/src/reality_capture/on_premise/_generic_manager.py new file mode 100644 index 00000000..69cf90ce --- /dev/null +++ b/python_sdk/src/reality_capture/on_premise/_generic_manager.py @@ -0,0 +1,233 @@ +import os +import sqlite3 +import time +from datetime import datetime +from typing import Optional + +from reality_capture.on_premise.result import Result, ManagerErrorCode + +# Constants + +## Version +DB_CURRENT_VERSION = "1.3" +JOB_CURRENT_VERSION = "2.0" + +## Table names +JOBS = "Jobs" +TASK_TABLE = "Tasks" +DEPS_TABLE = "Dependencies" +META = "Meta" + +## Jobs columns +JOBNAME = "JobName" +JOB_TYPE = "JobType" +STATUS = "Status" +PRIORITY = "Priority" +PERCENT = "Percent" +DESCR = "Descr" +SHARED_WORKING_DIR = "SharedWorkingDir" +SUBMIT_USER = "SubmitUser" +SUBMIT_HOST = "SubmitHost" +SUBMIT_TIME = "SubmitTime" +START_TIME = "StartTime" +END_TIME = "EndTime" +LAST_MSG = "LastMsg" +STEPS = "Steps" +CURRENT_STEP = "CurStep" +JOB_VERSION = "Version" +JOB_COMPUTE_VERSION = "ComputeVersion" + +## Tasks columns +TASKID = "TaskId" +STEP = "Step" +DEPTH = "Depth" +TYPE = "Type" +MAXNUMRUN = "MaxNumRun" +RETVAL = "RetVal" +NUMRUN = "NumRun" +WORKLOAD = "Workload" +RUN_HOST_NAME = "RunHostName" +RUN_USER_NAME = "RunUserName" +MILESTONE = "Milestone" + +## Dependencies columns +DEPENDENCYID = "DependencyID" + +## Meta columns +KEY = "Key" +VALUE = "Value" +DB_VERSION_KEY = "Version" + +def init_job_queue_db(job_queue_dir: str): + os.makedirs(job_queue_dir, exist_ok=True) + if os.path.exists(job_queue_dir + "/JobQueue.db"): + # We consider it was correctly created + return + create_new_jq(job_queue_dir) + return + +def create_new_jq(job_queue_dir: str): + """Create a brand-new JobQueue.db with the v1.3 schema.""" + db_path = job_queue_dir + "/JobQueue.db" + try: + conn = sqlite3.connect(db_path) + except sqlite3.OperationalError: + raise RuntimeError(f"Failed to create JobQueue database at '{db_path}'") + try: + cursor = conn.cursor() + + cursor.execute("PRAGMA foreign_keys=on;") + + # Jobs table + cursor.execute(f""" + CREATE TABLE IF NOT EXISTS {JOBS} ( + {JOBNAME} TEXT NOT NULL UNIQUE, + {STATUS} INT NOT NULL, + {PRIORITY} INT NOT NULL, + {PERCENT} REAL NOT NULL, + {DESCR} TEXT NOT NULL, + {SHARED_WORKING_DIR} TEXT, + {JOB_TYPE} TEXT, + {SUBMIT_USER} TEXT NOT NULL, + {SUBMIT_HOST} TEXT NOT NULL, + {SUBMIT_TIME} TEXT NOT NULL, + {START_TIME} TEXT, + {END_TIME} TEXT, + {LAST_MSG} TEXT, + {STEPS} TEXT, + {CURRENT_STEP} TEXT, + {JOB_VERSION} TEXT NOT NULL, + {JOB_COMPUTE_VERSION} TEXT NOT NULL + ); + """) + + # Tasks table + cursor.execute(f""" + CREATE TABLE IF NOT EXISTS {TASK_TABLE} ( + {JOBNAME} TEXT NOT NULL, + {TASKID} TEXT NOT NULL UNIQUE, + {STEP} TEXT, + {DEPTH} INTEGER, + {STATUS} INT NOT NULL, + {TYPE} INT NOT NULL, + {PERCENT} REAL NOT NULL, + {MAXNUMRUN} INT NOT NULL, + {RETVAL} INT, + {NUMRUN} INT NOT NULL, + {WORKLOAD} REAL NOT NULL, + {RUN_HOST_NAME} TEXT, + {RUN_USER_NAME} TEXT, + {START_TIME} TEXT, + {END_TIME} TEXT, + {LAST_MSG} TEXT, + {MILESTONE} TEXT + ); + """) + + # Dependencies table + cursor.execute(f""" + CREATE TABLE IF NOT EXISTS {DEPS_TABLE} ( + {JOBNAME} TEXT NOT NULL, + {TASKID} TEXT NOT NULL, + {DEPENDENCYID} TEXT + ); + """) + + # Meta table + cursor.execute(f""" + CREATE TABLE IF NOT EXISTS {META} ( + {KEY} TEXT NOT NULL UNIQUE, + {VALUE} TEXT NOT NULL + ); + """) + + # Insert current version + cursor.execute( + f"INSERT OR IGNORE INTO {META} ({KEY}, {VALUE}) VALUES (?, ?);", + (DB_VERSION_KEY, DB_CURRENT_VERSION) + ) + + # Indexes + cursor.execute(f"CREATE INDEX IF NOT EXISTS IdxJobs ON {JOBS} ({JOBNAME} ASC);") + cursor.execute(f"CREATE INDEX IF NOT EXISTS IdxTasks ON {TASK_TABLE} ({TASKID} ASC, {JOBNAME});") + cursor.execute(f"CREATE INDEX IF NOT EXISTS IdxJobsOnTasks ON {TASK_TABLE} ({JOBNAME} ASC);") + cursor.execute(f"CREATE INDEX IF NOT EXISTS IdxTasksJobsOnDeps ON {DEPS_TABLE} ({TASKID}, {JOBNAME});") + cursor.execute(f"CREATE INDEX IF NOT EXISTS IdxDeps ON {DEPS_TABLE} ({DEPENDENCYID});") + + conn.commit() + except Exception as e: + conn.rollback() + raise RuntimeError(f"Failed to init JobQueue database: {e}") from e + finally: + conn.close() + +def init_engine_db(job_queue_dir): + os.makedirs(job_queue_dir, exist_ok=True) + if os.path.exists(job_queue_dir + "/Engines.db"): + # We consider it was already correctly created + return + + db_path = job_queue_dir + "/Engines.db" + conn = sqlite3.connect(db_path) + try: + cursor = conn.cursor() + cursor.execute(""" + CREATE TABLE IF NOT EXISTS Engines ( + Edition TEXT NOT NULL, + Version TEXT NOT NULL, + Username TEXT NOT NULL, + Hostname TEXT NOT NULL, + Status INT NOT NULL, + StartTime TEXT NOT NULL, + LastHeartBeat TEXT NOT NULL, + EndTime TEXT, + Signal INT NOT NULL + ); + """) + conn.commit() + except Exception as e: + conn.rollback() + raise RuntimeError(f"Failed to create Engines database: {e}") from e + finally: + conn.close() + + + +class GenericManager: + def __init__(self, job_queue_dir: str): + self._connection = None + self._job_queue_dir = job_queue_dir + self._timeout_lock_s = 30 + init_job_queue_db(job_queue_dir) + init_engine_db(job_queue_dir) + + @staticmethod + def _acquire_lock(db_path: str, timeout: float) -> Optional[int]: + """Acquire a file-based lock (analogous to SemaphoreFile in C++).""" + lock_path = db_path + ".lock" + deadline = time.monotonic() + timeout + while True: + try: + fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + return fd + except FileExistsError: + if time.monotonic() >= deadline: + return None + time.sleep(0.05) + + @staticmethod + def _release_lock(fd: int, db_path: str): + """Release the file-based lock.""" + lock_path = db_path + ".lock" + os.close(fd) + os.remove(lock_path) + + @staticmethod + def _format_datetime(dt: datetime) -> str: + """Format datetime to string in format: YYYYMMDDTHHmmss.ffffff""" + return dt.strftime("%Y%m%dT%H%M%S.%f") + + @staticmethod + def _parse_datetime(dt_str: str) -> datetime: + """Parse datetime from string in format: YYYYMMDDTHHmmss.ffffff""" + return datetime.strptime(dt_str, "%Y%m%dT%H%M%S.%f") \ No newline at end of file diff --git a/python_sdk/src/reality_capture/on_premise/engine_manager.py b/python_sdk/src/reality_capture/on_premise/engine_manager.py new file mode 100644 index 00000000..7f84c5ec --- /dev/null +++ b/python_sdk/src/reality_capture/on_premise/engine_manager.py @@ -0,0 +1,205 @@ +from datetime import datetime +from typing import Optional +from pydantic import BaseModel, Field +from enum import Enum +import sqlite3 + +from reality_capture.on_premise._generic_manager import GenericManager +from reality_capture.on_premise.result import Result, ManagerErrorCode + + +class EngineSignal(Enum): + PAUSE = "Pause" + CLOSE = "Close" + FINISH = "Finish" + STOP = "Stop" + PAUSE_RIGHT_NOW = "PauseRightNow" + SKIP = "Skip" + UNPAUSE = "Unpause" + + +class EngineStatus(Enum): + UNKNOWN = "Unknown" + BUSY = "Busy" + READY = "Ready" + PAUSED = "Paused" + TURNED_OFF = "TurnedOff" + + +class EngineDetails(BaseModel): + host_name: str = Field(description="The hostname of the engine.") + user_name: str = Field(description="The user of the engine.") + version: str = Field(description="The version of the engine.") + start_time: datetime = Field(description="The start time of the engine.") + end_time: Optional[datetime] = Field(description="The end time of the engine if it was properly stopped.") + last_beat_time: Optional[datetime] = Field(description="The last beat time of the engine.") + status: EngineStatus = Field(description="The status of the engine.") + signal: Optional[list[EngineSignal]] = Field(description="The signals of the engine waiting to be processed.") + + +class EngineManager(GenericManager): + def __init__(self, job_queue_dir: str): + super().__init__(job_queue_dir) + self._db_path = self._job_queue_dir + "/Engines.db" + self._connection = sqlite3.connect(self._db_path) + + def _close(self): + if self._connection is not None: + self._connection.close() + self._connection = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self._close() + return False + + def __del__(self): + self._close() + + @staticmethod + def _int_to_status(status_as_int: int): + mapping = { + 1: EngineStatus.BUSY, + 2: EngineStatus.READY, + 4: EngineStatus.PAUSED, + 8: EngineStatus.TURNED_OFF, + } + if status_as_int in mapping: + return mapping[status_as_int] + return EngineStatus.UNKNOWN + + @staticmethod + def _int_to_signal(signal_as_int: int) -> list[EngineSignal]: + mapping = { + 1: EngineSignal.PAUSE, + 2: EngineSignal.CLOSE, + 4: EngineSignal.FINISH, + 8: EngineSignal.STOP, + 16: EngineSignal.PAUSE_RIGHT_NOW, + 32: EngineSignal.SKIP, + 64: EngineSignal.UNPAUSE, + } + signals = [] + for bit_value, signal in mapping.items(): + if signal_as_int & bit_value: + signals.append(signal) + return signals + + + @staticmethod + def _signal_to_int(signal: EngineSignal) -> int: + mapping = { + EngineSignal.PAUSE: 1, + EngineSignal.CLOSE: 2, + EngineSignal.FINISH: 4, + EngineSignal.STOP: 8, + EngineSignal.PAUSE_RIGHT_NOW: 16, + EngineSignal.SKIP: 32, + EngineSignal.UNPAUSE: 64, + } + return mapping[signal] + + def get_job_queue_dir(self) -> str: + """ + Return the job queue directory path. + + :return: The job queue directory path + """ + return self._job_queue_dir + + def get_engine_hostnames(self) -> Result[list[str]]: + """ + Get the list of engine hostnames in the JobQueue. Engines can be in any state. + + :return: A Result[list[str]] of engine hostnames + """ + try: + cursor = self._connection.cursor() + cursor.execute("SELECT Hostname FROM Engines") + engines = [row[0] for row in cursor.fetchall()] + except sqlite3.DatabaseError: + return Result(ManagerErrorCode.SQLITE_ERROR, None) + return Result(None, engines) + + def get_engine(self, engine_host_name: str) -> Result[EngineDetails]: + """ + Get the details for a specific engine + + :param engine_host_name: Name of the engine + :return: A Result[EngineDetails] with the engine details + """ + try: + cursor = self._connection.cursor() + cursor.execute( + "SELECT Version, Username, Hostname, Status, StartTime, LastHeartBeat, EndTime, Signal " + "FROM Engines WHERE Hostname = ?", + (engine_host_name,) + ) + row = cursor.fetchone() + + if row is None: + return Result(ManagerErrorCode.ENGINE_NOT_FOUND, None) + + version, username, hostname, status_int, start_time, last_beat_time, end_time, signal_int = row + + start_time = self._parse_datetime(start_time) + last_beat_time = self._parse_datetime(last_beat_time) if last_beat_time else None + end_time = self._parse_datetime(end_time) if end_time else None + eng_signal = self._int_to_signal(signal_int) if signal_int != 0 else None + + # Create EngineDetails, leaving status and signal as defaults + ed = EngineDetails( + host_name=hostname, + user_name=username, + version=version, + start_time=start_time, + end_time=end_time, + last_beat_time=last_beat_time, + signal=eng_signal, + status=self._int_to_status(status_int) + ) + except sqlite3.DatabaseError: + return Result(ManagerErrorCode.SQLITE_ERROR, None) + return Result(None, ed) + + def send_signal(self, engine_host_name: str, signal: EngineSignal) -> Result[EngineDetails]: + """ + Send a signal to an engine in order to change its behaviour + + :param engine_host_name: Name of the engine + :param signal: Signal to send to the engine + :return: A Result[EngineDetails] with the engine details + """ + lock_fd = self._acquire_lock(self._db_path, timeout=self._timeout_lock_s) + if lock_fd is None: + return Result(ManagerErrorCode.DB_BUSY, None) + try: + cursor = self._connection.cursor() + cursor.execute("BEGIN IMMEDIATE") + try: + # Check engine exists and is running + cursor.execute( + "SELECT COUNT(*) FROM Engines WHERE Hostname = ? AND (EndTime IS NULL OR EndTime = '')", + (engine_host_name,) + ) + count = cursor.fetchone()[0] + if count == 0: + self._connection.rollback() + return Result(ManagerErrorCode.ENGINE_NOT_FOUND, None) + + # Bitwise OR the new signal onto the existing signal + signal_int = self._signal_to_int(signal) + cursor.execute( + "UPDATE Engines SET Signal = Signal | ? WHERE Hostname = ? AND (EndTime IS NULL OR EndTime = '')", + (signal_int, engine_host_name) + ) + self._connection.commit() + except sqlite3.OperationalError: + self._connection.rollback() + return Result(ManagerErrorCode.SQLITE_ERROR, None) + finally: + self._release_lock(lock_fd, self._db_path) + + return self.get_engine(engine_host_name) diff --git a/python_sdk/src/reality_capture/on_premise/job.py b/python_sdk/src/reality_capture/on_premise/job.py new file mode 100644 index 00000000..5d44a911 --- /dev/null +++ b/python_sdk/src/reality_capture/on_premise/job.py @@ -0,0 +1,165 @@ +from pydantic import BaseModel, Field, ValidationInfo, field_validator +from datetime import datetime +from enum import Enum +from typing import Union, Any, Optional + +from reality_capture.common.job import BaseProgress, BaseExecution, JobState, JobType +from reality_capture.specifications.calibration import CalibrationSpecifications +from reality_capture.specifications.change_detection import ChangeDetectionSpecifications +from reality_capture.specifications.clearance import ClearanceSpecifications +from reality_capture.specifications.constraints import ConstraintsSpecifications +from reality_capture.specifications.fill_image_properties import FillImagePropertiesSpecifications +from reality_capture.specifications.import_point_cloud import ImportPCSpecifications +from reality_capture.specifications.objects2d import Objects2DSpecifications +from reality_capture.specifications.production import ProductionSpecifications +from reality_capture.specifications.reconstruction import ReconstructionSpecifications +from reality_capture.specifications.segmentation2d import Segmentation2DSpecifications +from reality_capture.specifications.segmentation3d import Segmentation3DSpecifications +from reality_capture.specifications.segmentation_orthophoto import SegmentationOrthophotoSpecifications +from reality_capture.specifications.tiling import TilingSpecifications +from reality_capture.specifications.touchup import (TouchUpImportSpecifications, TouchUpExportSpecifications) +from reality_capture.specifications.water_constraints import WaterConstraintsSpecifications +from reality_capture.specifications.gaussian_splats import GaussianSplatsSpecifications +from reality_capture.specifications.eval_o2d import EvalO2DSpecifications +from reality_capture.specifications.eval_o3d import EvalO3DSpecifications +from reality_capture.specifications.eval_s2d import EvalS2DSpecifications +from reality_capture.specifications.eval_s3d import EvalS3DSpecifications +from reality_capture.specifications.eval_sortho import EvalSOrthoSpecifications + + +class ActiveJob(BaseModel): + job_name: str = Field(description="Name of the job", alias="jobName") + running_tasks: int = Field(description="Number of running tasks", alias="runningTasks") + ready_tasks: int = Field(description="Number of tasks ready to be executed", alias="readyTasks") + + +class QueueSummary(BaseModel): + jobs_failed: int = Field(description="Number of failed jobs", alias="jobsFailed") + jobs_success: int = Field(description="Number of successful jobs", alias="jobsSuccess") + jobs_cancelled: int = Field(description="Number of cancelled jobs", alias="jobsCancelled") + jobs_active: list[ActiveJob] = Field(description="List of active jobs", alias="jobsActive") + jobs_queued: int = Field(description="Number of queued jobs", alias="jobsQueued") + + +class Milestone(BaseModel): + name: str = Field(description="Name of the milestone.") + parameters: list[str] = Field(default_factory=list, description="List of parameters.") + end_time: Optional[datetime] = Field(default=None, description="End time of the milestone.", alias="endTime") + + +class Progress(BaseProgress): + milestones: list[Milestone] = Field(description="State of the job.") + + +class ExecutionOnPrem(BaseExecution): + submit_host: str = Field(description="Computer who submitted the job.", alias="submitHost") + submit_user: str = Field(description="User who submitted the job.", alias="submitUser") + + +class JobPriority(Enum): + PAUSED = "Paused" + LOW = "Low" + NORMAL = "Normal" + HIGH = "High" + URGENT = "Urgent" + + +class JobFilters(BaseModel): + include_state: Optional[list[JobState]] = Field(default=None, description="Include job state", + alias="includeState") + created_date_time_range: Optional[tuple[datetime, datetime]] = Field(None, description="Select jobs created during this time range.", + alias="createdDateTimeRange") + ended_date_time_range: Optional[tuple[datetime, datetime]] = Field(None, description="Select jobs ended during this time range.", + alias="endedDateTimeRange") + started_date_time_range: Optional[tuple[datetime, datetime]] = Field(None, description="Select jobs started during this time range.", + alias="startedDateTimeRange") + limit: Optional[int] = Field(default=50, description="Number of jobs per page") + continuation_token: Optional[str] = Field(default=None, description="Continuation token to get the next page", + alias="continuationToken") + +class Job(BaseModel): + name: str = Field(description="Job name.") + priority: JobPriority = Field(description="Job priority.") + place: int = Field(description="Place of the job in the queue.") + processing_hosts: list[str] = Field(description="List of processing hosts for the job. " + "If running, these are the hosts running the job. " + "If ended, these are the hosts that executed at least one task for this job.", + alias="processingHosts") + state: JobState = Field(description="State of the job.") + execution_info: ExecutionOnPrem = Field(description="Known execution information for the job.", + alias="executionInfo") + type: JobType = Field(description="Type of the job.") + shared_working_dir: str = Field(description="Shared working directory for the job.", + alias="sharedWorkingDir") + specifications: Union[CalibrationSpecifications, ChangeDetectionSpecifications, ConstraintsSpecifications, + EvalO2DSpecifications, EvalO3DSpecifications, + EvalS2DSpecifications, EvalS3DSpecifications, + EvalSOrthoSpecifications, FillImagePropertiesSpecifications, + GaussianSplatsSpecifications, ImportPCSpecifications, + Objects2DSpecifications, ProductionSpecifications, + ReconstructionSpecifications, Segmentation2DSpecifications, + Segmentation3DSpecifications, SegmentationOrthophotoSpecifications, + TilingSpecifications, TouchUpExportSpecifications, + TouchUpImportSpecifications, WaterConstraintsSpecifications, + ClearanceSpecifications] = ( + Field(description="Specifications aligned with the job type.")) + + @field_validator("specifications", mode="plain") + @classmethod + def set_specification_validation_model(cls, raw_dict: dict[str, Any], validation_info: ValidationInfo): + job_type = validation_info.data['type'] + + specifications = None + + if job_type == JobType.CALIBRATION: + specifications = CalibrationSpecifications(**raw_dict) + elif job_type == JobType.CHANGE_DETECTION: + specifications = ChangeDetectionSpecifications(**raw_dict) + elif job_type == JobType.CONSTRAINTS: + specifications = ConstraintsSpecifications(**raw_dict) + elif job_type == JobType.EVAL_O2D: + specifications = EvalO2DSpecifications(**raw_dict) + elif job_type == JobType.EVAL_O3D: + specifications = EvalO3DSpecifications(**raw_dict) + elif job_type == JobType.EVAL_S2D: + specifications = EvalS2DSpecifications(**raw_dict) + elif job_type == JobType.EVAL_S3D: + specifications = EvalS3DSpecifications(**raw_dict) + elif job_type == JobType.EVAL_SORTHO: + specifications = EvalSOrthoSpecifications(**raw_dict) + elif job_type == JobType.FILL_IMAGE_PROPERTIES: + specifications = FillImagePropertiesSpecifications(**raw_dict) + elif job_type == JobType.GAUSSIAN_SPLATS: + specifications = GaussianSplatsSpecifications(**raw_dict) + elif job_type == JobType.IMPORT_POINT_CLOUD: + specifications = ImportPCSpecifications(**raw_dict) + elif job_type == JobType.OBJECTS_2D: + specifications = Objects2DSpecifications(**raw_dict) + elif job_type == JobType.PRODUCTION: + specifications = ProductionSpecifications(**raw_dict) + elif job_type == JobType.RECONSTRUCTION: + specifications = ReconstructionSpecifications(**raw_dict) + elif job_type == JobType.SEGMENTATION_2D: + specifications = Segmentation2DSpecifications(**raw_dict) + elif job_type == JobType.SEGMENTATION_3D: + specifications = Segmentation3DSpecifications(**raw_dict) + elif job_type == JobType.SEGMENTATION_ORTHOPHOTO: + specifications = SegmentationOrthophotoSpecifications(**raw_dict) + elif job_type == JobType.TILING: + specifications = TilingSpecifications(**raw_dict) + elif job_type == JobType.TOUCH_UP_EXPORT: + specifications = TouchUpExportSpecifications(**raw_dict) + elif job_type == JobType.TOUCH_UP_IMPORT: + specifications = TouchUpImportSpecifications(**raw_dict) + elif job_type == JobType.WATER_CONSTRAINTS: + specifications = WaterConstraintsSpecifications(**raw_dict) + elif job_type == JobType.CLEARANCE_CALCULATION: + specifications = ClearanceSpecifications(**raw_dict) + else: + raise ValueError(f"Unsupported job type: {job_type}") + + return specifications + +class JobPage(BaseModel): + jobs: list[Job] + next_continuation_token: Optional[str] = None \ No newline at end of file diff --git a/python_sdk/src/reality_capture/on_premise/job_manager.py b/python_sdk/src/reality_capture/on_premise/job_manager.py new file mode 100644 index 00000000..c82fa261 --- /dev/null +++ b/python_sdk/src/reality_capture/on_premise/job_manager.py @@ -0,0 +1,601 @@ +from datetime import datetime, timezone +from typing import Union, Optional +import uuid +import os +import json +import shutil +import socket +import getpass +import base64 +import sqlite3 + +import pydantic + +from reality_capture.on_premise._generic_manager import ( + GenericManager, + JOBS, JOBNAME, STATUS, PRIORITY, SHARED_WORKING_DIR, + JOB_TYPE, SUBMIT_USER, SUBMIT_HOST, SUBMIT_TIME, START_TIME, END_TIME, + PERCENT, TASK_TABLE, MILESTONE, RETVAL, DEPS_TABLE, + DESCR, LAST_MSG, STEPS, CURRENT_STEP, JOB_VERSION, JOB_COMPUTE_VERSION, + JOB_CURRENT_VERSION, +) + +from reality_capture.specifications.calibration import CalibrationSpecifications +from reality_capture.specifications.change_detection import ChangeDetectionSpecifications +from reality_capture.specifications.constraints import ConstraintsSpecifications +from reality_capture.specifications.fill_image_properties import FillImagePropertiesSpecifications +from reality_capture.specifications.import_point_cloud import ImportPCSpecifications +from reality_capture.specifications.objects2d import Objects2DSpecifications +from reality_capture.specifications.production import ProductionSpecifications +from reality_capture.specifications.reconstruction import ReconstructionSpecifications +from reality_capture.specifications.segmentation2d import Segmentation2DSpecifications +from reality_capture.specifications.segmentation3d import Segmentation3DSpecifications +from reality_capture.specifications.segmentation_orthophoto import SegmentationOrthophotoSpecifications +from reality_capture.specifications.tiling import TilingSpecifications +from reality_capture.specifications.touchup import (TouchUpImportSpecifications, TouchUpExportSpecifications) +from reality_capture.specifications.water_constraints import WaterConstraintsSpecifications +from reality_capture.specifications.gaussian_splats import GaussianSplatsSpecifications +from reality_capture.specifications.eval_o2d import EvalO2DSpecifications +from reality_capture.specifications.eval_o3d import EvalO3DSpecifications +from reality_capture.specifications.eval_s2d import EvalS2DSpecifications +from reality_capture.specifications.eval_s3d import EvalS3DSpecifications +from reality_capture.specifications.eval_sortho import EvalSOrthoSpecifications +from reality_capture.specifications.clearance import ClearanceSpecifications + +from reality_capture.common.job import JobState, JobType +from reality_capture.on_premise.job import (Job, JobPriority, ExecutionOnPrem, Progress, Milestone, JobFilters, + JobPage, QueueSummary, ActiveJob) +from reality_capture.on_premise.result import Result, ManagerErrorCode + +class JobManager(GenericManager): + def __init__(self, job_queue_dir: str): + super().__init__(job_queue_dir) + self._db_path = self._job_queue_dir + "/JobQueue.db" + self._connection = sqlite3.connect(self._db_path) + + def _close(self): + if self._connection is not None: + self._connection.close() + self._connection = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self._close() + return False + + def __del__(self): + self._close() + + @staticmethod + def _int_to_priority(jp: int) -> JobPriority: + mapping = { + -1: JobPriority.PAUSED, + 0: JobPriority.LOW, + 1: JobPriority.NORMAL, + 2: JobPriority.HIGH, + 3: JobPriority.URGENT, + } + return mapping[jp] + + @staticmethod + def _priority_to_int(jp: JobPriority) -> int: + mapping = { + JobPriority.PAUSED: -1, + JobPriority.LOW: 0, + JobPriority.NORMAL: 1, + JobPriority.HIGH: 2, + JobPriority.URGENT: 3, + } + return mapping.get(jp, 1) + + @staticmethod + def _int_to_state(si: int) -> JobState: + # C++ enum values (DB stores sum of status history, we want the highest) + # Pending=0x1, Running=0x4, Completed=0x8, Failed=0x10, Cancelled=0x20 + mapping = [ + (0x20, JobState.CANCELLED), + (0x10, JobState.FAILED), + (0x08, JobState.SUCCESS), + (0x04, JobState.ACTIVE), + (0x01, JobState.QUEUED), + ] + for flag, state in mapping: + if state == JobState.QUEUED: + continue + if si & flag: + return state + return JobState.QUEUED + + _STATE_SQL_MAP = { + JobState.QUEUED: f"({STATUS} & 1) AND NOT ({STATUS} & 60)", + JobState.ACTIVE: f"({STATUS} & 4) AND NOT ({STATUS} & 56)", + JobState.SUCCESS: f"({STATUS} & 8) AND NOT ({STATUS} & 48)", + JobState.FAILED: f"({STATUS} & 16) AND NOT ({STATUS} & 32)", + JobState.CANCELLED: f"({STATUS} & 32)", + } + + _SPECS_TYPE_TO_JOB_TYPE = { + CalibrationSpecifications: JobType.CALIBRATION, + ChangeDetectionSpecifications: JobType.CHANGE_DETECTION, + ConstraintsSpecifications: JobType.CONSTRAINTS, + EvalO2DSpecifications: JobType.EVAL_O2D, + EvalO3DSpecifications: JobType.EVAL_O3D, + EvalS2DSpecifications: JobType.EVAL_S2D, + EvalS3DSpecifications: JobType.EVAL_S3D, + EvalSOrthoSpecifications: JobType.EVAL_SORTHO, + FillImagePropertiesSpecifications: JobType.FILL_IMAGE_PROPERTIES, + GaussianSplatsSpecifications: JobType.GAUSSIAN_SPLATS, + ImportPCSpecifications: JobType.IMPORT_POINT_CLOUD, + Objects2DSpecifications: JobType.OBJECTS_2D, + ProductionSpecifications: JobType.PRODUCTION, + ReconstructionSpecifications: JobType.RECONSTRUCTION, + Segmentation2DSpecifications: JobType.SEGMENTATION_2D, + Segmentation3DSpecifications: JobType.SEGMENTATION_3D, + SegmentationOrthophotoSpecifications: JobType.SEGMENTATION_ORTHOPHOTO, + TilingSpecifications: JobType.TILING, + TouchUpExportSpecifications: JobType.TOUCH_UP_EXPORT, + TouchUpImportSpecifications: JobType.TOUCH_UP_IMPORT, + WaterConstraintsSpecifications: JobType.WATER_CONSTRAINTS, + ClearanceSpecifications: JobType.CLEARANCE_CALCULATION + } + + def _row_to_job_details(self, row) -> Result[Job]: + (name, status, priority, shared_working_dir, job_type, + submit_user, submit_host, submit_time, start_time, end_time) = row + + execution_info = ExecutionOnPrem( + submitUser=submit_user, + submitHost=submit_host or "", + createdDateTime=self._parse_datetime(submit_time), + startedDateTime=self._parse_datetime(start_time) if start_time else None, + endedDateTime=self._parse_datetime(end_time) if end_time else None, + ) + + # Load specifications from disk + # Format is jobqueue_dir/jobs/job_id/settings.json + settings_path = os.path.join(self._job_queue_dir, "jobs", name, "settings.json") + specs = {} + if os.path.exists(settings_path): + try: + with open(settings_path, "r", encoding="utf-8") as f: + specs = json.load(f) + except Exception: + return Result(ManagerErrorCode.CORRUPTED_SPECIFICATIONS, None) + else: + return Result(ManagerErrorCode.MISSING_SPECIFICATIONS, None) + + if not job_type: + return Result(ManagerErrorCode.INVALID_JOB_TYPE_IN_DB, None) + try: + job_type_enum = JobType(job_type) + except ValueError: + return Result(ManagerErrorCode.INVALID_JOB_TYPE_IN_DB, None) + + real_specs = specs.get(job_type) + if real_specs is None: + return Result(ManagerErrorCode.CORRUPTED_SPECIFICATIONS, None) + + try: + j = Job( + name=name, + priority=self._int_to_priority(priority), + place=0, + processingHosts=[], + state=self._int_to_state(status), + executionInfo=execution_info, + type=job_type_enum, + sharedWorkingDir=shared_working_dir or "", + specifications=real_specs, + ) + except pydantic.ValidationError: + return Result(ManagerErrorCode.CORRUPTED_SPECIFICATIONS, None) + + return Result(None, j) + + def get_job(self, job_name: str) -> Result[Job]: + """ + Retrieve job details from the JobQueue. + + :param job_name: Name of the job to retrieve. + :return: The job details retrieved from the JobQueue. + """ + cursor = self._connection.cursor() + cursor.execute( + f"SELECT {JOBNAME}, {STATUS}, {PRIORITY}, {SHARED_WORKING_DIR}, " + f"{JOB_TYPE}, {SUBMIT_USER}, {SUBMIT_HOST}, {SUBMIT_TIME}, " + f"{START_TIME}, {END_TIME} " + f"FROM {JOBS} WHERE {JOBNAME} = ?;", + (job_name,) + ) + row = cursor.fetchone() + if row is None: + return Result(ManagerErrorCode.JOB_NOT_FOUND, None) + + return self._row_to_job_details(row) + + def get_job_progress(self, job_name: str) -> Result[Progress]: + """ + Retrieve job progress from the JobQueue. + + :param job_name: Name of the job to retrieve. + :return: The job progress. + """ + cursor = self._connection.cursor() + + # Retrieve percentage and Status from the Jobs table + cursor.execute( + f"SELECT {PERCENT}, {STATUS} FROM {JOBS} WHERE {JOBNAME} = ?;", + (job_name,) + ) + job_row = cursor.fetchone() + if job_row is None: + return Result(ManagerErrorCode.JOB_NOT_FOUND, None) + + percent, status = job_row + state = self._int_to_state(status) + + # Retrieve Milestone, EndTime, and RetVal from the Tasks table + cursor.execute( + f"SELECT {MILESTONE}, {END_TIME}, {RETVAL} FROM {TASK_TABLE} WHERE {JOBNAME} = ?;", + (job_name,) + ) + task_rows = cursor.fetchall() + + milestones = [] + for milestone, end_time, ret_val in task_rows: + if milestone is None or "$" not in milestone: + continue + + # Milestone string is of format Name$Rank$Param1$Param2$... + # Params are optional and can be empty; Name and Rank are always there + splits = milestone.split("$") + m = Milestone(name=splits[0], parameters=splits[2:]) + m.end_time = self._parse_datetime(end_time) if end_time and ret_val == 0 else None + milestones.append(m) + + p = Progress( + state=state, + percentage=percent, + milestones=milestones + ) + return Result(None, p) + + def get_jobs(self, job_filters: JobFilters) -> Result[JobPage]: + """ + Retrieve jobs from the JobQueue based on specified job filters. Use a continuation token to get the next page of jobs. + + :param job_filters: Job filters to filter jobs on. + :return: The job page retrieved from the JobQueue. + """ + if job_filters.include_state is not None and len(job_filters.include_state) == 0: + return Result(None, JobPage(jobs=[], next_continuation_token=None)) + + where_clauses: list[str] = [] + params: list = [] + + # State filter + if job_filters.include_state is not None: + state_conditions = [ + f"({self._STATE_SQL_MAP[s]})" for s in job_filters.include_state + ] + where_clauses.append(f"({' OR '.join(state_conditions)})") + + # Datetime range filters + dt_range_mappings = [ + (job_filters.created_date_time_range, SUBMIT_TIME), + (job_filters.started_date_time_range, START_TIME), + (job_filters.ended_date_time_range, END_TIME), + ] + for dt_range, column in dt_range_mappings: + if dt_range is not None: + where_clauses.append(f"{column} BETWEEN ? AND ?") + params.append(self._format_datetime(dt_range[0])) + params.append(self._format_datetime(dt_range[1])) + + # Continuation token (keyset pagination via ROWID) + if job_filters.continuation_token is not None: + try: + last_rowid = int(base64.b64decode(job_filters.continuation_token).decode()) + except (UnicodeDecodeError, ValueError): + return Result(ManagerErrorCode.INVALID_CONTINUATION_TOKEN, None) + where_clauses.append("ROWID > ?") + params.append(last_rowid) + + # Build query + job_columns = ( + f"ROWID, {JOBNAME}, {STATUS}, {PRIORITY}, {SHARED_WORKING_DIR}, " + f"{JOB_TYPE}, {SUBMIT_USER}, {SUBMIT_HOST}, {SUBMIT_TIME}, " + f"{START_TIME}, {END_TIME}" + ) + query = f"SELECT {job_columns} FROM {JOBS}" + if where_clauses: + query += " WHERE " + " AND ".join(where_clauses) + query += " ORDER BY ROWID ASC LIMIT ?" + params.append(job_filters.limit + 1) + + cursor = self._connection.cursor() + cursor.execute(query, params) + rows = cursor.fetchall() + + # Determine if there's a next page + next_token: Optional[str] = None + if len(rows) > job_filters.limit: + rows = rows[:job_filters.limit] + last_rowid = rows[-1][0] # ROWID is the first column + next_token = base64.b64encode(str(last_rowid).encode()).decode() + + # Map rows to JobDetails (skip ROWID at index 0) + jobs_result = [self._row_to_job_details(row[1:]) for row in rows] + jobs: list[Job] = [] + for j in jobs_result: + if j.error is None and j.value is not None: + jobs.append(j.value) + return Result(None, JobPage(jobs=jobs, next_continuation_token=next_token)) + + def get_summary(self) -> Result[QueueSummary]: + """ + Produce a quick summary of the job queue, including counts of jobs in various states and a list of active jobs with their running and ready task counts. + + :return: The job queue summary. + """ + cursor = self._connection.cursor() + + # Count jobs by state + cursor.execute( + f"SELECT {STATUS} FROM {JOBS};" + ) + rows = cursor.fetchall() + + jobs_failed = 0 + jobs_success = 0 + jobs_cancelled = 0 + jobs_queued = 0 + + for (status,) in rows: + state = self._int_to_state(status) + if state == JobState.FAILED: + jobs_failed += 1 + elif state == JobState.SUCCESS: + jobs_success += 1 + elif state == JobState.CANCELLED: + jobs_cancelled += 1 + elif state == JobState.QUEUED: + jobs_queued += 1 + + # Get active jobs with their task counts + cursor.execute( + f"SELECT j.{JOBNAME}, t.{STATUS} FROM {JOBS} j " + f"INNER JOIN {TASK_TABLE} t ON j.{JOBNAME} = t.{JOBNAME} " + f"WHERE (j.{STATUS} & 4) AND NOT (j.{STATUS} & 56);" + ) + task_rows = cursor.fetchall() + + # Aggregate running/ready tasks per active job + active_jobs_map: dict[str, tuple[int, int]] = {} + for job_name, task_status in task_rows: + running, ready = active_jobs_map.get(job_name, (0, 0)) + # Running: bit 4 set, bits 8/16/32 not set + if (task_status & 4) and not (task_status & (8 | 16 | 32)): + running += 1 + # Ready: bit 2 set, bits 4/8/16/32 not set + elif (task_status & 2) and not (task_status & (4 | 8 | 16 | 32)): + ready += 1 + active_jobs_map[job_name] = (running, ready) + + jobs_active = [ + ActiveJob(jobName=name, runningTasks=running, readyTasks=ready) + for name, (running, ready) in active_jobs_map.items() + ] + qs = QueueSummary( + jobsFailed=jobs_failed, + jobsSuccess=jobs_success, + jobsCancelled=jobs_cancelled, + jobsQueued=jobs_queued, + jobsActive=jobs_active, + ) + + return Result(None, qs) + + def submit_job(self, specifications: Union[CalibrationSpecifications, ChangeDetectionSpecifications, ConstraintsSpecifications, + EvalO2DSpecifications, EvalO3DSpecifications, + EvalS2DSpecifications, EvalS3DSpecifications, + EvalSOrthoSpecifications, FillImagePropertiesSpecifications, + GaussianSplatsSpecifications, ImportPCSpecifications, + Objects2DSpecifications, ProductionSpecifications, + ReconstructionSpecifications, Segmentation2DSpecifications, + Segmentation3DSpecifications, SegmentationOrthophotoSpecifications, + TilingSpecifications, TouchUpExportSpecifications, + TouchUpImportSpecifications, WaterConstraintsSpecifications, + ClearanceSpecifications], + shared_working_directory: str, + priority: JobPriority = JobPriority.NORMAL, workspace: Optional[str] = None) -> Result[Job]: + """ + Submit a job to the job queue. + + :param specifications: The specifications of the job + :param shared_working_directory: The shared working directory unique to this job + :param priority: The priority of the job + :param workspace: The workspace to leverage for the job. + :return: The job details of the submitted job. + """ + if not shared_working_directory: + return Result(ManagerErrorCode.EMPTY_SHARED_WORKING_DIRECTORY, None) + fd = self._acquire_lock(self._db_path, timeout=self._timeout_lock_s) + if fd is None: + return Result(ManagerErrorCode.DB_BUSY, None) + try: + cursor = self._connection.cursor() + cursor.execute("BEGIN IMMEDIATE") + + job_name = f"job_{uuid.uuid4()}" + submit_time = self._format_datetime(datetime.now(timezone.utc)) + submit_user = getpass.getuser() + submit_host = socket.gethostname() + + # Derive job type from specifications type + job_type_enum = self._SPECS_TYPE_TO_JOB_TYPE.get(type(specifications)) + if job_type_enum is None: + self._connection.rollback() + return Result(ManagerErrorCode.UNSUPPORTED_SPECIFICATIONS, None) + job_type = job_type_enum.value + + specs_dict = specifications.model_dump(by_alias=True, exclude_none=True) + specifications_payload = {job_type: specs_dict} + + # Save specifications to disk + settings_dir = os.path.join(self._job_queue_dir, "jobs", job_name) + os.makedirs(settings_dir, exist_ok=True) + settings_path = os.path.join(settings_dir, "settings.json") + with open(settings_path, "w", encoding="utf-8") as f: + if workspace is not None and workspace != "": + specifications_payload[job_type].setdefault("options", {})["workspace"] = workspace + json.dump(specifications_payload, f, indent=2) + + # Insert job into database + try: + cursor.execute( + f"INSERT INTO {JOBS} (" + f"{JOBNAME}, {STATUS}, {PRIORITY}, {PERCENT}, {DESCR}, " + f"{SHARED_WORKING_DIR}, {JOB_TYPE}, {SUBMIT_USER}, {SUBMIT_HOST}, " + f"{SUBMIT_TIME}, {START_TIME}, {END_TIME}, {LAST_MSG}, " + f"{STEPS}, {CURRENT_STEP}, {JOB_VERSION}, {JOB_COMPUTE_VERSION}" + f") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);", + ( + job_name, + 1, # Pending status + self._priority_to_int(priority), + 0.0, + "", # description + shared_working_directory, + job_type, + submit_user, + submit_host, + submit_time, + None, # start_time + None, # end_time + None, # last_msg + "Prepare_Job", # steps + None, # current_step + JOB_CURRENT_VERSION, + "", # compute_version + ) + ) + self._connection.commit() + except Exception: + shutil.rmtree(settings_dir, ignore_errors=True) + self._connection.rollback() + return Result(ManagerErrorCode.SQLITE_ERROR, None) + finally: + self._release_lock(fd, self._db_path) + + return self.get_job(job_name) + + def set_job_priority(self, job_name: str, job_priority: JobPriority) -> Result[Job]: + """ + Update the priority of the job + + :param job_name: The name of the job + :param job_priority: The priority of the job + :return: The job details of the updated job + """ + fd = self._acquire_lock(self._db_path, timeout=self._timeout_lock_s) + if fd is None: + return Result(ManagerErrorCode.DB_BUSY, None) + try: + cursor = self._connection.cursor() + cursor.execute("BEGIN IMMEDIATE") + + try: + cursor.execute( + f"UPDATE {JOBS} SET {PRIORITY} = ? WHERE {JOBNAME} = ?;", + (self._priority_to_int(job_priority), job_name) + ) + if cursor.rowcount != 1: + self._connection.rollback() + return Result(ManagerErrorCode.JOB_NOT_FOUND, None) + self._connection.commit() + except Exception: + self._connection.rollback() + return Result(ManagerErrorCode.SQLITE_ERROR, None) + finally: + self._release_lock(fd, self._db_path) + + return self.get_job(job_name) + + def cancel_job(self, job_name: str) -> Result[Job]: + """ + Cancel a specific job + + :param job_name: The name of the job to cancel + :return: The job details of the cancelled job + """ + cancelled_mask = 0x20 # 32 + final_state_mask = 0x38 # 56 = Completed(0x08) | Failed(0x10) | Cancelled(0x20) + + fd = self._acquire_lock(self._db_path, timeout=self._timeout_lock_s) + if fd is None: + return Result(ManagerErrorCode.DB_BUSY, None) + try: + cursor = self._connection.cursor() + cursor.execute("BEGIN IMMEDIATE") + + try: + # Fetch current status and shared working directory + cursor.execute( + f"SELECT {STATUS}, {SHARED_WORKING_DIR} FROM {JOBS} WHERE {JOBNAME} = ?;", + (job_name,) + ) + row = cursor.fetchone() + if row is None: + self._connection.rollback() + return Result(ManagerErrorCode.JOB_NOT_FOUND, None) + + status, shared_working_dir = row + shared_working_dir = shared_working_dir or "" + + # Validate: only pending (0x1) or pending+running (0x5) jobs can be cancelled + if status & final_state_mask: + self._connection.rollback() + return Result(ManagerErrorCode.JOB_NOT_CANCELLABLE, None) + + cancel_time = self._format_datetime(datetime.now(timezone.utc)) + + # Update job status, end time, and last message + cursor.execute( + f"UPDATE {JOBS} SET " + f"{STATUS} = {STATUS} | {cancelled_mask}, " + f"{END_TIME} = ?, " + f"{LAST_MSG} = ? " + f"WHERE {JOBNAME} = ?;", + (cancel_time, "Job has been cancelled", job_name) + ) + + # Cancel not-finished tasks (tasks without any final state bit set) + cursor.execute( + f"UPDATE {TASK_TABLE} SET " + f"{STATUS} = {STATUS} | {cancelled_mask}, " + f"{END_TIME} = ? " + f"WHERE {JOBNAME} = ? AND ({STATUS} & {final_state_mask}) = 0;", + (cancel_time, job_name) + ) + + # Delete dependencies for this job + cursor.execute( + f"DELETE FROM {DEPS_TABLE} WHERE {JOBNAME} = ?;", + (job_name,) + ) + + self._connection.commit() + except Exception: + self._connection.rollback() + return Result(ManagerErrorCode.SQLITE_ERROR, None) + finally: + self._release_lock(fd, self._db_path) + + # Delete shared working directory (after lock release, non-reversible) + if shared_working_dir: + shutil.rmtree(shared_working_dir, ignore_errors=True) + + return self.get_job(job_name) diff --git a/python_sdk/src/reality_capture/on_premise/result.py b/python_sdk/src/reality_capture/on_premise/result.py new file mode 100644 index 00000000..ce94510a --- /dev/null +++ b/python_sdk/src/reality_capture/on_premise/result.py @@ -0,0 +1,75 @@ +from typing import TypeVar, Generic, Optional +from enum import Enum + + +T = TypeVar("T") + + +class ManagerErrorCode(Enum): + JOB_NOT_FOUND = "JobNotFound" + ENGINE_NOT_FOUND = "EngineNotFound" + SQLITE_ERROR = "SqliteError" + CORRUPTED_SPECIFICATIONS = "CorruptedSpecifications" + MISSING_SPECIFICATIONS = "MissingSpecifications" + INVALID_JOB_TYPE_IN_DB = "InvalidJobTypeInDB" + EMPTY_SHARED_WORKING_DIRECTORY = "EmptySharedWorkingDirectory" + UNSUPPORTED_SPECIFICATIONS = "UnsupportedSpecifications" + DB_BUSY = "DBBusy" + JOB_NOT_CANCELLABLE = "JobNotCancellable" + INVALID_CONTINUATION_TOKEN = "InvalidContinuationToken" + + +class Result(tuple, Generic[T]): + """ + A tuple containing a ManagerErrorCode or a value. + """ + + error: Optional[ManagerErrorCode] + "Optional error if the request failed." + value: Optional[T] + "Optional object if the request succeed." + + def __new__(cls, error: Optional[ManagerErrorCode], value: Optional[T]): + self = tuple.__new__(cls, (error, value)) + self.value = value + self.error = error + return self + + def is_error(self) -> bool: + """ + Checks whether the response is an error response. + + :return: True if the response contains a valid error. + """ + return self.error is not None + + def get_error_as_str(self) -> str: + """ + Get the error message. + + :return: The error message. + """ + match self.error: + case ManagerErrorCode.JOB_NOT_FOUND: + return "Job not found." + case ManagerErrorCode.ENGINE_NOT_FOUND: + return "Engine not found." + case ManagerErrorCode.SQLITE_ERROR: + return "Sqlite error. If the issue persists, raise an issue on the RealityCapture GitHub repository." + case ManagerErrorCode.CORRUPTED_SPECIFICATIONS: + return "Corrupted specifications. If the issue persists, raise an issue on the RealityCapture GitHub repository." + case ManagerErrorCode.MISSING_SPECIFICATIONS: + return "Missing specifications in database folder." + case ManagerErrorCode.INVALID_JOB_TYPE_IN_DB: + return "Invalid job type in the database." + case ManagerErrorCode.EMPTY_SHARED_WORKING_DIRECTORY: + return "Empty shared working directory." + case ManagerErrorCode.UNSUPPORTED_SPECIFICATIONS: + return "Unsupported specification." + case ManagerErrorCode.DB_BUSY: + return "DB is busy, try again later." + case ManagerErrorCode.JOB_NOT_CANCELLABLE: + return "Job can't be cancelled due to its current state." + case ManagerErrorCode.INVALID_CONTINUATION_TOKEN: + return "Invalid continuation token." + return "" diff --git a/python_sdk/src/reality_capture/service/job.py b/python_sdk/src/reality_capture/service/job.py index 2c095918..a7b8fd4a 100644 --- a/python_sdk/src/reality_capture/service/job.py +++ b/python_sdk/src/reality_capture/service/job.py @@ -1,8 +1,9 @@ import urllib.parse from pydantic import BaseModel, Field, ValidationInfo, field_validator -from datetime import datetime from enum import Enum from typing import Union, Optional, Any + +from reality_capture.common.job import JobType, JobState, BaseProgress, BaseExecution from reality_capture.specifications.calibration import CalibrationSpecifications, CalibrationSpecificationsCreate from reality_capture.specifications.change_detection import (ChangeDetectionSpecifications, ChangeDetectionSpecificationsCreate) @@ -41,30 +42,6 @@ from reality_capture.service.reality_data import URL -class JobType(Enum): - CALIBRATION = "Calibration" - CHANGE_DETECTION = "ChangeDetection" - CONSTRAINTS = "Constraints" - EVAL_O2D = "EvalO2D" - EVAL_O3D = "EvalO3D" - EVAL_S2D = "EvalS2D" - EVAL_S3D = "EvalS3D" - EVAL_SORTHO = "EvalSOrtho" - FILL_IMAGE_PROPERTIES = "FillImageProperties" - GAUSSIAN_SPLATS = "GaussianSplats" - IMPORT_POINT_CLOUD = "ImportPointCloud" - OBJECTS_2D = "Objects2D" - PRODUCTION = "Production" - RECONSTRUCTION = "Reconstruction" - SEGMENTATION_2D = "Segmentation2D" - SEGMENTATION_3D = "Segmentation3D" - SEGMENTATION_ORTHOPHOTO = "SegmentationOrthophoto" - TILING = "Tiling" - TOUCH_UP_IMPORT = "TouchUpImport" - TOUCH_UP_EXPORT = "TouchUpExport" - WATER_CONSTRAINTS = "WaterConstraints" - CLEARANCE_CALCULATION = "ClearanceCalculation" - # POINT_CLOUD_CONVERSION = "PointCloudConversion" class Service(Enum): MODELING = "Modeling" @@ -74,43 +51,33 @@ class Service(Enum): def _get_appropriate_service(jt: JobType): if jt in [JobType.FILL_IMAGE_PROPERTIES, JobType.IMPORT_POINT_CLOUD, JobType.CALIBRATION, JobType.TILING, - JobType.PRODUCTION, JobType.RECONSTRUCTION, JobType.CONSTRAINTS, JobType.TOUCH_UP_EXPORT, - JobType.TOUCH_UP_IMPORT, JobType.WATER_CONSTRAINTS, JobType.GAUSSIAN_SPLATS]: + JobType.PRODUCTION, JobType.RECONSTRUCTION, JobType.CONSTRAINTS, JobType.TOUCH_UP_EXPORT, + JobType.TOUCH_UP_IMPORT, JobType.WATER_CONSTRAINTS, JobType.GAUSSIAN_SPLATS]: return Service.MODELING if jt in [JobType.OBJECTS_2D, JobType.SEGMENTATION_2D, JobType.SEGMENTATION_3D, JobType.SEGMENTATION_ORTHOPHOTO, - JobType.CHANGE_DETECTION, JobType.EVAL_O2D, JobType.EVAL_O3D, JobType.EVAL_S2D, - JobType.EVAL_S3D, JobType.EVAL_SORTHO, JobType.CLEARANCE_CALCULATION]: + JobType.CHANGE_DETECTION, JobType.EVAL_O2D, JobType.EVAL_O3D, JobType.EVAL_S2D, + JobType.EVAL_S3D, JobType.EVAL_SORTHO, JobType.CLEARANCE_CALCULATION]: return Service.ANALYSIS # return Service.CONVERSION raise NotImplementedError("Other services not yet implemented") -class JobState(Enum): - QUEUED = "Queued" - ACTIVE = "Active" - SUCCESS = "Success" - FAILED = "Failed" - TERMINATING_ON_CANCEL = "TerminatingOnCancel" - TERMINATING_ON_FAILURE = "TerminatingOnFailure" - CANCELLED = "Cancelled" - - class JobCreate(BaseModel): name: Optional[str] = Field(None, description="Displayable job name.", min_length=3) type: JobType = Field(description="Type of job.") # TODO : PointCloudConversionSpecificationsCreate, specifications: Union[CalibrationSpecificationsCreate, ChangeDetectionSpecificationsCreate, - ConstraintsSpecificationsCreate, - EvalO2DSpecificationsCreate, EvalO3DSpecificationsCreate, - EvalS2DSpecificationsCreate, EvalS3DSpecificationsCreate, - EvalSOrthoSpecificationsCreate, FillImagePropertiesSpecificationsCreate, - GaussianSplatsSpecificationsCreate, ImportPCSpecificationsCreate, - Objects2DSpecificationsCreate, ProductionSpecificationsCreate, - ReconstructionSpecificationsCreate, Segmentation2DSpecificationsCreate, - Segmentation3DSpecificationsCreate, SegmentationOrthophotoSpecificationsCreate, - TilingSpecificationsCreate, TouchUpExportSpecificationsCreate, - TouchUpImportSpecificationsCreate, WaterConstraintsSpecificationsCreate, - ClearanceSpecificationsCreate] = ( + ConstraintsSpecificationsCreate, + EvalO2DSpecificationsCreate, EvalO3DSpecificationsCreate, + EvalS2DSpecificationsCreate, EvalS3DSpecificationsCreate, + EvalSOrthoSpecificationsCreate, FillImagePropertiesSpecificationsCreate, + GaussianSplatsSpecificationsCreate, ImportPCSpecificationsCreate, + Objects2DSpecificationsCreate, ProductionSpecificationsCreate, + ReconstructionSpecificationsCreate, Segmentation2DSpecificationsCreate, + Segmentation3DSpecificationsCreate, SegmentationOrthophotoSpecificationsCreate, + TilingSpecificationsCreate, TouchUpExportSpecificationsCreate, + TouchUpImportSpecificationsCreate, WaterConstraintsSpecificationsCreate, + ClearanceSpecificationsCreate] = ( Field(description="Specifications aligned with the job type.")) itwin_id: str = Field(description="iTwin ID, used by the service for finding " "input reality data and uploading output data.", @@ -125,12 +92,9 @@ def get_appropriate_service(self) -> Service: return _get_appropriate_service(self.type) -class Execution(BaseModel): - created_date_time: datetime = Field(description="Creation date time for the job.", alias="createdDateTime") - started_date_time: Optional[datetime] = Field(None, description="Start date time for the job.", alias="startedDateTime") - ended_date_time: Optional[datetime] = Field(None, description="End date time for the job.", alias="endedDateTime") +class Execution(BaseExecution): processing_units: Optional[float] = Field(None, description="Processing units consumed by the job.", - alias="processingUnits") + alias="processingUnits") class Job(BaseModel): @@ -145,17 +109,17 @@ class Job(BaseModel): user_id: str = Field(description="Identifier of the user that created the job.", alias="userId") # TODO : add PointCloudConversionSpecifications specifications: Union[CalibrationSpecifications, ChangeDetectionSpecifications, - ConstraintsSpecifications, - EvalO2DSpecifications, EvalO3DSpecifications, - EvalS2DSpecifications, EvalS3DSpecifications, - EvalSOrthoSpecifications, FillImagePropertiesSpecifications, - GaussianSplatsSpecifications, ImportPCSpecifications, - Objects2DSpecifications, ProductionSpecifications, - ReconstructionSpecifications, Segmentation2DSpecifications, - Segmentation3DSpecifications, SegmentationOrthophotoSpecifications, - TilingSpecifications, TouchUpExportSpecifications, - TouchUpImportSpecifications, WaterConstraintsSpecifications, - ClearanceSpecifications] = ( + ConstraintsSpecifications, + EvalO2DSpecifications, EvalO3DSpecifications, + EvalS2DSpecifications, EvalS3DSpecifications, + EvalSOrthoSpecifications, FillImagePropertiesSpecifications, + GaussianSplatsSpecifications, ImportPCSpecifications, + Objects2DSpecifications, ProductionSpecifications, + ReconstructionSpecifications, Segmentation2DSpecifications, + Segmentation3DSpecifications, SegmentationOrthophotoSpecifications, + TilingSpecifications, TouchUpExportSpecifications, + TouchUpImportSpecifications, WaterConstraintsSpecifications, + ClearanceSpecifications] = ( Field(description="Specifications aligned with the job type.")) @field_validator("specifications", mode="plain") @@ -172,15 +136,15 @@ def set_specification_validation_model(cls, raw_dict: dict[str, Any], validation elif job_type == JobType.CONSTRAINTS: specifications = ConstraintsSpecifications(**raw_dict) elif job_type == JobType.EVAL_O2D: - specifications = EvalO2DSpecifications(**raw_dict) + specifications = EvalO2DSpecifications(**raw_dict) elif job_type == JobType.EVAL_O3D: - specifications = EvalO3DSpecifications(**raw_dict) + specifications = EvalO3DSpecifications(**raw_dict) elif job_type == JobType.EVAL_S2D: - specifications = EvalS2DSpecifications(**raw_dict) + specifications = EvalS2DSpecifications(**raw_dict) elif job_type == JobType.EVAL_S3D: - specifications = EvalS3DSpecifications(**raw_dict) + specifications = EvalS3DSpecifications(**raw_dict) elif job_type == JobType.EVAL_SORTHO: - specifications = EvalSOrthoSpecifications(**raw_dict) + specifications = EvalSOrthoSpecifications(**raw_dict) elif job_type == JobType.FILL_IMAGE_PROPERTIES: specifications = FillImagePropertiesSpecifications(**raw_dict) elif job_type == JobType.GAUSSIAN_SPLATS: @@ -194,11 +158,11 @@ def set_specification_validation_model(cls, raw_dict: dict[str, Any], validation elif job_type == JobType.RECONSTRUCTION: specifications = ReconstructionSpecifications(**raw_dict) elif job_type == JobType.SEGMENTATION_2D: - specifications = Segmentation2DSpecifications(**raw_dict) + specifications = Segmentation2DSpecifications(**raw_dict) elif job_type == JobType.SEGMENTATION_3D: - specifications = Segmentation3DSpecifications(**raw_dict) + specifications = Segmentation3DSpecifications(**raw_dict) elif job_type == JobType.SEGMENTATION_ORTHOPHOTO: - specifications = SegmentationOrthophotoSpecifications(**raw_dict) + specifications = SegmentationOrthophotoSpecifications(**raw_dict) elif job_type == JobType.TILING: specifications = TilingSpecifications(**raw_dict) elif job_type == JobType.TOUCH_UP_EXPORT: @@ -253,9 +217,8 @@ def get_continuation_token(self) -> Optional[str]: return cts[0] -class Progress(BaseModel): - state: JobState = Field(description="State of the job.") - percentage: float = Field(ge=0, le=100, description="Progress of the job.") +class Progress(BaseProgress): + pass class ProgressResponse(BaseModel): diff --git a/python_sdk/src/reality_capture/specifications/calibration.py b/python_sdk/src/reality_capture/specifications/calibration.py index ff58f5e8..a68e753d 100644 --- a/python_sdk/src/reality_capture/specifications/calibration.py +++ b/python_sdk/src/reality_capture/specifications/calibration.py @@ -6,14 +6,12 @@ class CalibrationInputs(BaseModel): scene: str = Field(description="Reality data ID of ContextScene to process") presets: Optional[list[str]] = Field(default=None, description="List of paths to preset") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class CalibrationOutputs(BaseModel): scene: str = Field(description="Reality data ID of calibrated ContextScene") - report: Optional[str] = Field(default=None, description="Path in the bucket of Calibration report", - pattern=r"^bkt:.+") + report: Optional[str] = Field(default=None, description="Path in the bucket of Calibration report") textured_tie_points: Optional[str] = Field(default=None, description="Reality data ID of textured tie points", alias="texturedTiePoints") diff --git a/python_sdk/src/reality_capture/specifications/change_detection.py b/python_sdk/src/reality_capture/specifications/change_detection.py index eec92092..62ad9559 100644 --- a/python_sdk/src/reality_capture/specifications/change_detection.py +++ b/python_sdk/src/reality_capture/specifications/change_detection.py @@ -6,7 +6,7 @@ class ChangeDetectionInputs(BaseModel): model_3d_a: str = Field(alias="model3dA", description="Reality data id of ContextScene, point cloud or mesh") model_3d_b: str = Field(alias="model3dB", description="Reality data id of ContextScene, point cloud or mesh") - extent: Optional[str] = Field(None, alias="extent", pattern=r"^bkt:.+", + extent: Optional[str] = Field(None, alias="extent", description="Path in the bucket of the clipping polygon to apply") diff --git a/python_sdk/src/reality_capture/specifications/constraints.py b/python_sdk/src/reality_capture/specifications/constraints.py index 096a5f8b..297aeaed 100644 --- a/python_sdk/src/reality_capture/specifications/constraints.py +++ b/python_sdk/src/reality_capture/specifications/constraints.py @@ -9,13 +9,11 @@ class ConstraintType(Enum): class ConstraintToAdd(BaseModel): - constraint_path: str = Field(alias="constraintPath", description="Path in the bucket to the constraint file", - pattern=r"^bkt:.+") + constraint_path: str = Field(alias="constraintPath", description="Path in the bucket to the constraint file") crs: str = Field(description="Coordinate reference system") type: Optional[ConstraintType] = Field(None, description="Type of the constraint") resolution: Optional[float] = Field(None, description="Resolution of the constraint") - texture_path: Optional[str] = Field(None, alias="texturePath", description="Path in the bucket to the texture file", - pattern=r"^bkt:.+") + texture_path: Optional[str] = Field(None, alias="texturePath", description="Path in the bucket to the texture file") texture_size: Optional[int] = Field(None, alias="textureSize", description="Size of the texture") fill_color: Optional[str] = Field(None, alias="fillColor", description="Fill color for the constraint") name: Optional[str] = Field(None, description="Name of the constraint") @@ -38,14 +36,12 @@ class ConstraintsInputs(BaseModel): description="IDs of constraints to delete") constraints_to_add: Optional[list[ConstraintToAdd]] = Field(None, alias="constraintsToAdd", description="Constraints to add") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class ConstraintsOutputs(BaseModel): added_constraints_info: str = Field(alias="addedConstraintsInfo", - description="Path in the bucket for added ConstraintsInfo", - pattern=r"^bkt:.+") + description="Path in the bucket for added ConstraintsInfo",) class ConstraintsOutputsCreate(Enum): diff --git a/python_sdk/src/reality_capture/specifications/eval_o2d.py b/python_sdk/src/reality_capture/specifications/eval_o2d.py index 656c7a7b..cf171225 100644 --- a/python_sdk/src/reality_capture/specifications/eval_o2d.py +++ b/python_sdk/src/reality_capture/specifications/eval_o2d.py @@ -10,8 +10,7 @@ class EvalO2DInputs(BaseModel): class EvalO2DOutputs(BaseModel): - report: Optional[str] = Field(None, description="Path in Bucket of json report with binary classification.", - pattern=r"^bkt:.+") + report: Optional[str] = Field(None, description="Path in Bucket of json report with binary classification.") objects2d: Optional[str] = Field(None, alias="objects2D", description="Reality data id of ContextScene, " "annotated with classified embedded 2D objects") diff --git a/python_sdk/src/reality_capture/specifications/eval_o3d.py b/python_sdk/src/reality_capture/specifications/eval_o3d.py index e7e9b11f..f4eca967 100644 --- a/python_sdk/src/reality_capture/specifications/eval_o3d.py +++ b/python_sdk/src/reality_capture/specifications/eval_o3d.py @@ -10,8 +10,7 @@ class EvalO3DInputs(BaseModel): class EvalO3DOutputs(BaseModel): - report: Optional[str] = Field(None, description="Path in Bucket of json report with binary classification", - pattern=r"^bkt:.+") + report: Optional[str] = Field(None, description="Path in Bucket of json report with binary classification") objects3d: Optional[str] = Field(None, alias="objects3D", description="Reality data id of ContextScene, " "annotated with classified embedded 3D objects") diff --git a/python_sdk/src/reality_capture/specifications/eval_s2d.py b/python_sdk/src/reality_capture/specifications/eval_s2d.py index 057176d4..0647660a 100644 --- a/python_sdk/src/reality_capture/specifications/eval_s2d.py +++ b/python_sdk/src/reality_capture/specifications/eval_s2d.py @@ -11,8 +11,7 @@ class EvalS2DInputs(BaseModel): class EvalS2DOutputs(BaseModel): - report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix", - pattern=r"^bkt:.+") + report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix") segmented_photos: Optional[str] = Field(None, alias="segmentedPhotos", description="Reality data id of segmented photos, " "annotated with confusion matrix index") diff --git a/python_sdk/src/reality_capture/specifications/eval_s3d.py b/python_sdk/src/reality_capture/specifications/eval_s3d.py index 9a5de6ed..a6948ef9 100644 --- a/python_sdk/src/reality_capture/specifications/eval_s3d.py +++ b/python_sdk/src/reality_capture/specifications/eval_s3d.py @@ -11,8 +11,7 @@ class EvalS3DInputs(BaseModel): class EvalS3DOutputs(BaseModel): - report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix", - pattern=r"^bkt:.+") + report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix") segmented_point_cloud: Optional[str] = Field(None, alias="segmentedPointCloud", description="Reality data id of segmented point cloud, " "annotated with confusion matrix index") diff --git a/python_sdk/src/reality_capture/specifications/eval_sortho.py b/python_sdk/src/reality_capture/specifications/eval_sortho.py index d637a9d0..00542fb4 100644 --- a/python_sdk/src/reality_capture/specifications/eval_sortho.py +++ b/python_sdk/src/reality_capture/specifications/eval_sortho.py @@ -11,8 +11,7 @@ class EvalSOrthoInputs(BaseModel): class EvalSOrthoOutputs(BaseModel): - report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix", - pattern=r"^bkt:.+") + report: Optional[str] = Field(None, description="Path in Bucket of json report with confusion matrix") segmented_photos: Optional[str] = Field(None, alias="segmentedPhotos", description="Reality data id of segmented photos, " "annotated with confusion matrix index") diff --git a/python_sdk/src/reality_capture/specifications/gaussian_splats.py b/python_sdk/src/reality_capture/specifications/gaussian_splats.py index b95283f9..2ab281bc 100644 --- a/python_sdk/src/reality_capture/specifications/gaussian_splats.py +++ b/python_sdk/src/reality_capture/specifications/gaussian_splats.py @@ -10,10 +10,8 @@ class GaussianSplatsInputs(BaseModel): alias="splatsReference") region_of_interest: Optional[str] = Field(description="Path in the bucket to region of interest file", alias="regionOfInterest", - default=None, - pattern=r"^bkt:.+") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + default=None) + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") preset: Optional[str] = Field(default=None, description="Path to preset") diff --git a/python_sdk/src/reality_capture/specifications/import_point_cloud.py b/python_sdk/src/reality_capture/specifications/import_point_cloud.py index 2bce3227..a2bfe1fa 100644 --- a/python_sdk/src/reality_capture/specifications/import_point_cloud.py +++ b/python_sdk/src/reality_capture/specifications/import_point_cloud.py @@ -6,8 +6,7 @@ class ImportPCInputs(BaseModel): scene: str = Field(description="Reality data id of ContextScene to process") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class ImportPCOutputs(BaseModel): diff --git a/python_sdk/src/reality_capture/specifications/production.py b/python_sdk/src/reality_capture/specifications/production.py index 6d9389ca..3133eb5f 100644 --- a/python_sdk/src/reality_capture/specifications/production.py +++ b/python_sdk/src/reality_capture/specifications/production.py @@ -10,11 +10,9 @@ class ProductionInputs(BaseModel): modeling_reference: str = Field(description="Reality data id of modeling reference to process", alias="modelingReference") extent: Optional[str] = Field(None, description="Path in the bucket to region of interest file, " - "used for export extent", - pattern=r"^bkt:.+") + "used for export extent") presets: Optional[list[str]] = Field(default=None, description="List of paths to preset") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class Format(Enum): diff --git a/python_sdk/src/reality_capture/specifications/reconstruction.py b/python_sdk/src/reality_capture/specifications/reconstruction.py index c1b3965e..066ef06b 100644 --- a/python_sdk/src/reality_capture/specifications/reconstruction.py +++ b/python_sdk/src/reality_capture/specifications/reconstruction.py @@ -8,16 +8,13 @@ class ReconstructionInputs(BaseModel): scene: str = Field(description="Reality data id of ContextScene to process") region_of_interest: Optional[str] = Field(description="Path in the bucket to region of interest file, " "used for tiling region of interest", - pattern=r"^bkt:.+", alias="regionOfInterest", default=None) extent: Optional[str] = Field(None, description="Path in the bucket to region of interest file, " - "used for export extent", - pattern=r"^bkt:.+") + "used for export extent") modeling_reference: Optional[str] = Field(None, description="Reality data id of modeling reference to process", alias="modelingReference") presets: Optional[list[str]] = Field(default=None, description="List of paths to preset") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class ReconstructionOutputs(BaseModel): diff --git a/python_sdk/src/reality_capture/specifications/segmentation3d.py b/python_sdk/src/reality_capture/specifications/segmentation3d.py index f50fd04a..8c9ab6cb 100644 --- a/python_sdk/src/reality_capture/specifications/segmentation3d.py +++ b/python_sdk/src/reality_capture/specifications/segmentation3d.py @@ -19,7 +19,7 @@ class Segmentation3DInputs(BaseModel): "pointing to a segmented point cloud, " "this input replaces point_cloud_segmentation_detector, " "point_clouds and meshes inputs") - extent: Optional[str] = Field(None, alias="extent", pattern=r"^bkt:.+", + extent: Optional[str] = Field(None, alias="extent", description="Path in the bucket of the clipping polygon to apply") diff --git a/python_sdk/src/reality_capture/specifications/tiling.py b/python_sdk/src/reality_capture/specifications/tiling.py index a86cbfa8..97d7fa9a 100644 --- a/python_sdk/src/reality_capture/specifications/tiling.py +++ b/python_sdk/src/reality_capture/specifications/tiling.py @@ -8,11 +8,9 @@ class TilingInputs(BaseModel): scene: str = Field(description="Reality data id of ContextScene to process") region_of_interest: Optional[str] = Field(description="Path in the bucket to region of interest file", alias="regionOfInterest", - default=None, - pattern=r"^bkt:.+") + default=None) presets: Optional[list[str]] = Field(default=None, description="List of paths to preset") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class ModelingReferenceType(Enum): diff --git a/python_sdk/src/reality_capture/specifications/touchup.py b/python_sdk/src/reality_capture/specifications/touchup.py index 6b0d7fdc..5bb25d14 100644 --- a/python_sdk/src/reality_capture/specifications/touchup.py +++ b/python_sdk/src/reality_capture/specifications/touchup.py @@ -54,8 +54,7 @@ class TouchUpImportOutputsCreate(Enum): class TouchUpImportOutputs(BaseModel): import_info: Optional[str] = Field(None, alias="importInfo", description="Folder in bucket containing the " - "information about what was imported", - pattern=r"^bkt:.+") + "information about what was imported") class TouchUpImportSpecificationsCreate(BaseModel): diff --git a/python_sdk/src/reality_capture/specifications/water_constraints.py b/python_sdk/src/reality_capture/specifications/water_constraints.py index ebdde203..3f3adccc 100644 --- a/python_sdk/src/reality_capture/specifications/water_constraints.py +++ b/python_sdk/src/reality_capture/specifications/water_constraints.py @@ -6,8 +6,7 @@ class WaterConstraintsInputs(BaseModel): scene: str = Field(description="Reality data id of ContextScene") modeling_reference: str = Field(alias="modelingReference", description="Reality data id of Modeling Reference") - crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData", - pattern=r"^bkt:.+") + crs_data: Optional[str] = Field(default=None, description="Path in the bucket for CRS data.", alias="crsData") class WaterConstraintsOptions(BaseModel): @@ -20,8 +19,7 @@ class WaterConstraintsOutputsCreate(Enum): class WaterConstraintsOutputs(BaseModel): - constraints: str = Field(description="Path in the bucket of output constraints", - pattern=r"^bkt:.+") + constraints: str = Field(description="Path in the bucket of output constraints") class WaterConstraintsSpecificationsCreate(BaseModel): diff --git a/python_sdk/tests/data/DB_Engines/Engines.db b/python_sdk/tests/data/DB_Engines/Engines.db new file mode 100644 index 00000000..75de43b1 --- /dev/null +++ b/python_sdk/tests/data/DB_Engines/Engines.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:dbeb2828f21e10efe099eacffdaa779ae3686c8e175ea3b51285e0bf7ce3c1e6 +size 8192 diff --git a/python_sdk/tests/data/DB_Engines/JobQueue.db b/python_sdk/tests/data/DB_Engines/JobQueue.db new file mode 100644 index 00000000..18fd2651 --- /dev/null +++ b/python_sdk/tests/data/DB_Engines/JobQueue.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:31705db76787318a7a97f8b0aa2a0e76c9ec17524d638bcc60e049bb536e894c +size 61440 diff --git a/python_sdk/tests/data/DB_Jobs/Engines.db b/python_sdk/tests/data/DB_Jobs/Engines.db new file mode 100644 index 00000000..97b79b4b --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/Engines.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:a19297e92711182d46988f97f73d3be03d9d4edfbd30bd28229659c0704d7c84 +size 8192 diff --git a/python_sdk/tests/data/DB_Jobs/JobQueue.db b/python_sdk/tests/data/DB_Jobs/JobQueue.db new file mode 100644 index 00000000..43653b8e --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/JobQueue.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:041faa8cd6045bf903b4d8be1972b5626b5fdb348407af44e1511a4ba3704b53 +size 114688 diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json new file mode 100644 index 00000000..7947f739 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_3/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_3/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_3", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_3/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_6448e60c-a22a-44f7-819e-1a71068feebd/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_6448e60c-a22a-44f7-819e-1a71068feebd/settings.json new file mode 100644 index 00000000..d7862e5c --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_6448e60c-a22a-44f7-819e-1a71068feebd/settings.json @@ -0,0 +1,31 @@ +{ + "Production" : { + "inputs" : { + "modelingReference" : "D:/Projects/TestDB/Project files/Block_2/Reconstruction_1", + "scene" : "D:/Projects/TestDB/Project files/Block_2/Reconstructions-Internal/Reconstruction_1/Production_2/tmp_inputs/blockScene", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "exports" : [ + { + "format" : "3DTiles", + "location" : "D:/Projects/TestDB/Productions/Production_2", + "name" : "Production_2", + "options" : { + "textureColorSource" : "Visible", + "textureColorSourceResMin" : 0.0761929999999999968, + "textureColorSourceResMax" : 0.118620000000000003, + "textureColorSourceThermalUnit" : "Celsius", + "textureColorSourceThermalMin" : -1, + "textureColorSourceThermalMax" : -1, + "crs" : "EPSG:4978", + "lodScope" : "AcrossTiles" + } + } + ] + }, + "options" : { + "workspace" : "D:/Projects/TestDB/Project files/Block_2/ProductionWorkspace" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json new file mode 100644 index 00000000..1c55e36b --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_6/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_6/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_6", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_6/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : true, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "None", + "rigidRegistrationRotation" : "None", + "rigidRegistrationScale" : "None", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json new file mode 100644 index 00000000..79dcbc7e --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json @@ -0,0 +1,31 @@ +{ + "Production" : { + "inputs" : { + "modelingReference" : "D:/Projects/TestDB/Project files/Block_2/Reconstruction_1", + "scene" : "D:/Projects/TestDB/Project files/Block_2/Reconstructions-Internal/Reconstruction_1/Production_1/tmp_inputs/blockScene", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "exports" : [ + { + "format" : "3DTiles", + "location" : "D:/Projects/TestDB/Productions/Production_1", + "name" : "Production_1", + "options" : { + "textureColorSource" : "Visible", + "textureColorSourceResMin" : 0.0761929999999999968, + "textureColorSourceResMax" : 0.118620000000000003, + "textureColorSourceThermalUnit" : "Celsius", + "textureColorSourceThermalMin" : -1, + "textureColorSourceThermalMax" : -1, + "crs" : "EPSG:4978", + "lodScope" : "AcrossTiles" + } + } + ] + }, + "options" : { + "workspace" : "D:/Projects/TestDB/Project files/Block_2/ProductionWorkspace" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json new file mode 100644 index 00000000..107b3e16 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_2/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_2/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_2", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_2/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json b/python_sdk/tests/data/DB_Jobs/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json new file mode 100644 index 00000000..45b4f94c --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_4/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_4/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_4", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_4/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/Engines.db b/python_sdk/tests/data/DB_Jobs_Corrupted/Engines.db new file mode 100644 index 00000000..97b79b4b --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/Engines.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:a19297e92711182d46988f97f73d3be03d9d4edfbd30bd28229659c0704d7c84 +size 8192 diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/JobQueue.db b/python_sdk/tests/data/DB_Jobs_Corrupted/JobQueue.db new file mode 100644 index 00000000..64b40cb8 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/JobQueue.db @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:a3da081d58f72f6777e22a21a9078ad7cadda6c18ead11805f94c03e857cb60f +size 114688 diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json new file mode 100644 index 00000000..7947f739 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_12083864-05aa-4a5a-b183-6242fec8bfa5/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_3/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_3/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_3", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_3/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json new file mode 100644 index 00000000..21ddb796 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6/settings.json @@ -0,0 +1 @@ +This is not a json \ No newline at end of file diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json new file mode 100644 index 00000000..79dcbc7e --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78/settings.json @@ -0,0 +1,31 @@ +{ + "Production" : { + "inputs" : { + "modelingReference" : "D:/Projects/TestDB/Project files/Block_2/Reconstruction_1", + "scene" : "D:/Projects/TestDB/Project files/Block_2/Reconstructions-Internal/Reconstruction_1/Production_1/tmp_inputs/blockScene", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "exports" : [ + { + "format" : "3DTiles", + "location" : "D:/Projects/TestDB/Productions/Production_1", + "name" : "Production_1", + "options" : { + "textureColorSource" : "Visible", + "textureColorSourceResMin" : 0.0761929999999999968, + "textureColorSourceResMax" : 0.118620000000000003, + "textureColorSourceThermalUnit" : "Celsius", + "textureColorSourceThermalMin" : -1, + "textureColorSourceThermalMax" : -1, + "crs" : "EPSG:4978", + "lodScope" : "AcrossTiles" + } + } + ] + }, + "options" : { + "workspace" : "D:/Projects/TestDB/Project files/Block_2/ProductionWorkspace" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json new file mode 100644 index 00000000..107b3e16 --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "scene" : "D:/Projects/TestDB/Project files/Block_2/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "report" : "D:/Projects/TestDB/Project files/Block_2/QualityReport", + "scene" : "D:/Projects/TestDB/Project files/Block_2", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_2/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json new file mode 100644 index 00000000..5bd63e4f --- /dev/null +++ b/python_sdk/tests/data/DB_Jobs_Corrupted/jobs/job_faa0be51-3f34-452b-a4ea-885ebe9015b3/settings.json @@ -0,0 +1,39 @@ +{ + "Calibration" : { + "inputs" : { + "s" : "D:/Projects/TestDB/Project files/Block_4/AT-Internal/Scene0", + "crsData" : "D:/Projects/TestDB/Project files/CRSData" + }, + "outputs" : { + "r" : "D:/Projects/TestDB/Project files/Block_4/QualityReport", + "s" : "D:/Projects/TestDB/Project files/Block_4", + "texturedTiePoints" : "D:/Projects/TestDB/Project files/Block_4/Splats" + }, + "options" : { + "rigSynchro" : "None", + "rotationPolicy" : "Compute", + "centerPolicy" : "Compute", + "focalPolicy" : "Adjust", + "principalPolicy" : "Adjust", + "radialPolicy" : "Adjust", + "tangentialPolicy" : "Adjust", + "fisheyeFocalPolicy" : "Keep", + "fisheyeDistortionPolicy" : "Keep", + "aspectRatioPolicy" : "Keep", + "skewPolicy" : "Keep", + "tiepointsPolicy" : "Compute", + "pairSelection" : "Default", + "pairSelectionDistance" : 3, + "keypointsDensity" : "Normal", + "precalibration" : false, + "colorEqualization" : "BlockWise", + "adjustmentConstraints" : [ + "None" + ], + "rigidRegistrationPosition" : "PositionMetadata", + "rigidRegistrationRotation" : "PositionMetadata", + "rigidRegistrationScale" : "PositionMetadata", + "workspace" : "D:/Projects/TestDB/Project files" + } + } +} diff --git a/python_sdk/tests/test_onprem_engine.py b/python_sdk/tests/test_onprem_engine.py new file mode 100644 index 00000000..abf64ea0 --- /dev/null +++ b/python_sdk/tests/test_onprem_engine.py @@ -0,0 +1,133 @@ +import shutil +import stat +import pytest +import os + +from reality_capture.on_premise.engine_manager import EngineManager, EngineStatus, EngineSignal +from reality_capture.on_premise.result import ManagerErrorCode + + +class TestOnPremEngine: + @pytest.fixture(autouse=True) + def tmp_folder(self, tmp_path): + self.tmp_dir = str(tmp_path) + yield + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def test_create(self): + em = EngineManager(self.tmp_dir + "/jq") + assert em.get_job_queue_dir() == self.tmp_dir + "/jq" + hostnames_res = em.get_engine_hostnames() + assert not hostnames_res.is_error() + assert hostnames_res.value is not None + assert len(hostnames_res.value) == 0 + + get_result = em.get_engine("COMPUTER") + assert get_result.is_error() + assert get_result.value is None + assert get_result.error is ManagerErrorCode.ENGINE_NOT_FOUND + + def test_with(self): + with EngineManager(self.tmp_dir + "/jq") as em: + assert em.get_job_queue_dir() == self.tmp_dir + "/jq" + + def test_faulty_dbs(self): + # Create a faulty database file + os.mkdir(self.tmp_dir + "/jq") + faulty_db_path = self.tmp_dir + "/jq/Engines.db" + with open(faulty_db_path, "w") as f: + f.write("This is not a valid database file.") + + em = EngineManager(self.tmp_dir + "/jq") + res = em.get_engine_hostnames() + assert res.is_error() + assert res.value is None + assert res.error is ManagerErrorCode.SQLITE_ERROR + + res = em.get_engine("COMPUTER") + assert res.is_error() + assert res.value is None + assert res.error is ManagerErrorCode.SQLITE_ERROR + + def test_existing_db(self): + # Copy data from DB_Engines to tmp dir + current_dir = os.path.dirname(os.path.abspath(__file__)) + source_db_path = os.path.join(current_dir, "data", "DB_Engines", "Engines.db") + target_db_path = os.path.join(self.tmp_dir, "jq", "Engines.db") + os.makedirs(os.path.dirname(target_db_path), exist_ok=True) + shutil.copyfile(source_db_path, target_db_path) + + em = EngineManager(self.tmp_dir + "/jq") + res = em.get_engine_hostnames() + assert not res.is_error() + assert res.value is not None + assert len(res.value) == 2 + engine_name = res.value[0] + assert engine_name == "MINITEL" + + res = em.get_engine(engine_name) + assert not res.is_error() + assert res.value is not None + engine = res.value + assert engine.host_name == "MINITEL" + assert engine.version == "26.0.3.99999" + assert engine.user_name == "Rene.Coty" + assert engine.status == EngineStatus.READY + assert engine.start_time.isoformat() == "2026-06-26T08:15:12.473267" + assert engine.end_time is None + assert engine.last_beat_time is not None + assert engine.last_beat_time.isoformat() == "2026-06-26T08:15:56.032184" + assert engine.signal is None + + res = em.send_signal(engine_name, EngineSignal.PAUSE) + assert not res.is_error() + assert res.value is not None + engine = res.value + assert engine.signal is not None + assert engine.signal == [EngineSignal.PAUSE] + + res = em.send_signal("TO8", EngineSignal.PAUSE) + assert res.is_error() + assert res.error == ManagerErrorCode.ENGINE_NOT_FOUND + + res = em.get_engine("TO7") + assert not res.is_error() + assert res.value is not None + assert res.value.status == EngineStatus.UNKNOWN + + def test_existing_db_is_read_only(self): + # Copy data from DB_Engines to tmp dir + current_dir = os.path.dirname(os.path.abspath(__file__)) + source_db_path = os.path.join(current_dir, "data", "DB_Engines", "Engines.db") + target_db_path = os.path.join(self.tmp_dir, "jq", "Engines.db") + os.makedirs(os.path.dirname(target_db_path), exist_ok=True) + shutil.copyfile(source_db_path, target_db_path) + + # Make target_db_path read only + os.chmod(target_db_path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) + + em = EngineManager(self.tmp_dir + "/jq") + res = em.send_signal("MINITEL", EngineSignal.PAUSE) + assert res.is_error() + assert res.error == ManagerErrorCode.SQLITE_ERROR + + def test_db_is_locked(self): + # Copy data from DB_Engines to tmp dir + current_dir = os.path.dirname(os.path.abspath(__file__)) + source_db_path = os.path.join(current_dir, "data", "DB_Engines", "Engines.db") + target_db_path = os.path.join(self.tmp_dir, "jq", "Engines.db") + os.makedirs(os.path.dirname(target_db_path), exist_ok=True) + shutil.copyfile(source_db_path, target_db_path) + + em = EngineManager(self.tmp_dir + "/jq") + em2 = EngineManager(self.tmp_dir + "/jq") + + fd = em._acquire_lock(target_db_path, 5) + assert fd is not None + try: + em2._timeout_lock_s = 3 # For speed's sake + res = em2.send_signal("MINITEL", EngineSignal.PAUSE) + assert res.is_error() + assert res.error == ManagerErrorCode.DB_BUSY + finally: + em._release_lock(fd, target_db_path) \ No newline at end of file diff --git a/python_sdk/tests/test_onprem_generic_manager.py b/python_sdk/tests/test_onprem_generic_manager.py new file mode 100644 index 00000000..c1f84ece --- /dev/null +++ b/python_sdk/tests/test_onprem_generic_manager.py @@ -0,0 +1,87 @@ +import shutil +import sqlite3 +from os.path import exists +from unittest.mock import patch, MagicMock + +import pytest +import os +import sys + +from reality_capture.on_premise.engine_manager import EngineManager +from reality_capture.on_premise.job_manager import JobManager +from reality_capture.on_premise.result import ManagerErrorCode, Result + + +class TestOnPremEngine: + @pytest.fixture(autouse=True) + def tmp_folder(self, tmp_path): + self.tmp_dir = str(tmp_path) + yield + # Make sure directory is writable before cleanup (for read-only test) + for root, dirs, files in os.walk(self.tmp_dir): + for d in dirs: + try: + dirpath = os.path.join(root, d) + if sys.platform == "win32": + import subprocess + subprocess.run(["icacls", dirpath, "/remove:d", "Everyone"], capture_output=True) + else: + os.chmod(dirpath, 0o755) + except (OSError, PermissionError): + pass + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def test_read_only_folder(self): + # Create a read-only directory + os.makedirs(self.tmp_dir, exist_ok=True) + jq_path = os.path.join(self.tmp_dir, "jq") + os.mkdir(jq_path) + + # Set read-only permissions (cross-platform) + if sys.platform == "win32": + import subprocess + subprocess.run(["icacls", jq_path, "/deny", "Everyone:(W,D,AD)"], check=True, capture_output=True) + else: + os.chmod(jq_path, 0o555) + + # Attempt to create EngineManager in a read-only directory + with pytest.raises(RuntimeError): + jm = JobManager(jq_path) + + def test_error_message(self): + for e in ManagerErrorCode: + jde = Result(e, None) + assert jde.get_error_as_str() != "" + + a = 2 + jde = Result(None, a) + assert jde.get_error_as_str() == "" + + def test_init_throw_job(self): + jq_path = os.path.join(self.tmp_dir, "jq") + os.mkdir(jq_path) + + mock_cursor = MagicMock() + mock_cursor.execute.side_effect = [None, sqlite3.OperationalError("DB error")] + + mock_conn = MagicMock() + mock_conn.cursor.return_value = mock_cursor + + with patch("sqlite3.connect", return_value=mock_conn): + with pytest.raises(RuntimeError): + jm = JobManager(jq_path) + + def test_init_throw_engine(self): + jq_path = os.path.join(self.tmp_dir, "jq") + os.mkdir(jq_path) + open(os.path.join(jq_path, "JobQueue.db"), "w").close() + + mock_cursor = MagicMock() + mock_cursor.execute.side_effect = [sqlite3.OperationalError("DB error")] + + mock_conn = MagicMock() + mock_conn.cursor.return_value = mock_cursor + + with patch("sqlite3.connect", return_value=mock_conn): + with pytest.raises(RuntimeError): + em = EngineManager(jq_path) \ No newline at end of file diff --git a/python_sdk/tests/test_onprem_job_validator.py b/python_sdk/tests/test_onprem_job_validator.py new file mode 100644 index 00000000..b5372808 --- /dev/null +++ b/python_sdk/tests/test_onprem_job_validator.py @@ -0,0 +1,375 @@ +from reality_capture.on_premise.job import Job +from reality_capture.specifications.calibration import CalibrationSpecifications +from reality_capture.specifications.change_detection import ChangeDetectionSpecifications +from reality_capture.specifications.constraints import ConstraintsSpecifications +from reality_capture.specifications.eval_o2d import EvalO2DSpecifications +from reality_capture.specifications.eval_o3d import EvalO3DSpecifications +from reality_capture.specifications.eval_s2d import EvalS2DSpecifications +from reality_capture.specifications.eval_s3d import EvalS3DSpecifications +from reality_capture.specifications.eval_sortho import EvalSOrthoSpecifications +from reality_capture.specifications.fill_image_properties import FillImagePropertiesSpecifications +from reality_capture.specifications.gaussian_splats import GaussianSplatsSpecifications +from reality_capture.specifications.import_point_cloud import ImportPCSpecifications +from reality_capture.specifications.objects2d import Objects2DSpecifications +from reality_capture.specifications.production import ProductionSpecifications +from reality_capture.specifications.reconstruction import ReconstructionSpecifications +from reality_capture.specifications.segmentation2d import Segmentation2DSpecifications +from reality_capture.specifications.segmentation3d import Segmentation3DSpecifications +from reality_capture.specifications.segmentation_orthophoto import SegmentationOrthophotoSpecifications +from reality_capture.specifications.tiling import TilingSpecifications +from reality_capture.specifications.touchup import TouchUpImportSpecifications, TouchUpExportSpecifications +from reality_capture.specifications.water_constraints import WaterConstraintsSpecifications +from reality_capture.specifications.clearance import ClearanceSpecifications +import pytest +from unittest.mock import patch, MagicMock + +class TestOnPremJobValidator: + j_base = { + "name": "test", + "priority": "Normal", + "state": "Queued", + "place": 3, + "processingHosts": [], + "executionInfo": { + "createdDateTime": "2025-01-19T14:30:00Z", + "submitHost": "MINITEL", + "submitUser": "Rene.Coty" + }, + "sharedWorkingDir": "dir" + } + + def test_validation_fip(self): + j = self.j_base.copy() + j["type"] = "FillImageProperties" + j["specifications"] = { + "inputs": { + "imageCollections": ["ic_id"] + }, + "outputs": { + "scene": "sceneid" + } + } + job = Job(**j) + assert isinstance(job.specifications, FillImagePropertiesSpecifications) + + def test_validation_calibration(self): + j = self.j_base.copy() + j["type"] = "Calibration" + j["specifications"] = { + "inputs": { + "scene": "scene_id" + }, + "outputs": { + "scene": "sceneid" + } + } + job = Job(**j) + assert isinstance(job.specifications, CalibrationSpecifications) + + def test_validation_change_detection(self): + j = self.j_base.copy() + j["type"] = "ChangeDetection" + j["specifications"] = { + "inputs": { + "model3dA": "modela", + "model3dB": "modelb" + }, + "outputs": { + "changesInModelA": "rdid", + "objects3d": "obj" + } + } + job = Job(**j) + assert isinstance(job.specifications, ChangeDetectionSpecifications) + + def test_validation_constraints(self): + j = self.j_base.copy() + j["type"] = "Constraints" + j["specifications"] = { + "inputs": { + "modelingReference": "mfid", + "constraints_to_delete": ["4161f47e-24b4-4f97-802e-d68b71bdcb65"] + }, + "outputs": { + "addedConstraintsInfo": "bkt:youpi/t.json" + } + } + job = Job(**j) + assert isinstance(job.specifications, ConstraintsSpecifications) + + def test_validation_eval_o2d(self): + j = self.j_base.copy() + j["type"] = "EvalO2D" + j["specifications"] = { + "inputs": { + "reference": "mfid", + "prediction": "predid" + }, + "outputs": { + "objects2d": "objid" + } + } + job = Job(**j) + assert isinstance(job.specifications, EvalO2DSpecifications) + + def test_validation_eval_o3d(self): + j = self.j_base.copy() + j["type"] = "EvalO3D" + j["specifications"] = { + "inputs": { + "reference": "mfid", + "prediction": "predid" + }, + "outputs": { + "objects3d": "objid" + } + } + job = Job(**j) + assert isinstance(job.specifications, EvalO3DSpecifications) + + def test_validation_eval_s2d(self): + j = self.j_base.copy() + j["type"] = "EvalS2D" + j["specifications"] = { + "inputs": { + "reference": "mfid", + "prediction": "predid" + }, + "outputs": { + "segmentation2d": "sid" + } + } + job = Job(**j) + assert isinstance(job.specifications, EvalS2DSpecifications) + + def test_validation_eval_s3d(self): + j = self.j_base.copy() + j["type"] = "EvalS3D" + j["specifications"] = { + "inputs": { + "reference": "mfid", + "prediction": "predid" + }, + "outputs": { + "segmentation3d": "sid" + } + } + job = Job(**j) + assert isinstance(job.specifications, EvalS3DSpecifications) + + def test_validation_eval_sortho(self): + j = self.j_base.copy() + j["type"] = "EvalSOrtho" + j["specifications"] = { + "inputs": { + "reference": "mfid", + "prediction": "predid" + }, + "outputs": { + "segmentation2d": "sid" + } + } + job = Job(**j) + assert isinstance(job.specifications, EvalSOrthoSpecifications) + + def test_validation_gs(self): + j = self.j_base.copy() + j["type"] = "GaussianSplats" + j["specifications"] = { + "inputs": { + "scene": "mfid" + }, + "outputs": { + "splats": "sid" + } + } + job = Job(**j) + assert isinstance(job.specifications, GaussianSplatsSpecifications) + + def test_validation_ipc(self): + j = self.j_base.copy() + j["type"] = "ImportPointCloud" + j["specifications"] = { + "inputs": { + "scene": "mfid" + }, + "outputs": { + "scene": "sid", + "scanCollection": "sid2" + } + } + job = Job(**j) + assert isinstance(job.specifications, ImportPCSpecifications) + + def test_validation_o2d(self): + j = self.j_base.copy() + j["type"] = "Objects2D" + j["specifications"] = { + "inputs": { + "photos": "mfid" + }, + "outputs": { + "objects2d": "sid" + } + } + job = Job(**j) + assert isinstance(job.specifications, Objects2DSpecifications) + + def test_validation_prod(self): + j = self.j_base.copy() + j["type"] = "Production" + j["specifications"] = { + "inputs": { + "scene": "mfid", + "modelingReference": "rmid" + }, + "outputs": { + "exports": [{"location": "eid", "format": "LAS"}] + } + } + job = Job(**j) + assert isinstance(job.specifications, ProductionSpecifications) + + def test_validation_recons(self): + j = self.j_base.copy() + j["type"] = "Reconstruction" + j["specifications"] = { + "inputs": { + "scene": "mfid", + }, + "outputs": { + "exports": [{"location": "eid", "format": "LAS"}] + } + } + job = Job(**j) + assert isinstance(job.specifications, ReconstructionSpecifications) + + def test_validation_s2d(self): + j = self.j_base.copy() + j["type"] = "Segmentation2D" + j["specifications"] = { + "inputs": { + "photos": "mfid", + }, + "outputs": { + "segmentation2d": "s2did" + } + } + job = Job(**j) + assert isinstance(job.specifications, Segmentation2DSpecifications) + + def test_validation_s3d(self): + j = self.j_base.copy() + j["type"] = "Segmentation3D" + j["specifications"] = { + "inputs": { + "model3d": "mfid", + }, + "outputs": { + "segmentation3D": "s3did" + } + } + job = Job(**j) + assert isinstance(job.specifications, Segmentation3DSpecifications) + + def test_validation_sortho(self): + j = self.j_base.copy() + j["type"] = "SegmentationOrthophoto" + j["specifications"] = { + "inputs": { + "orthophoto": "mfid", + "orthophotoSegmentationDetector": "detector" + }, + "outputs": { + "segmentation2d": "s2did" + } + } + job = Job(**j) + assert isinstance(job.specifications, SegmentationOrthophotoSpecifications) + + def test_validation_tiling(self): + j = self.j_base.copy() + j["type"] = "Tiling" + j["specifications"] = { + "inputs": { + "scene": "mfid" + }, + "outputs": { + "modelingReference": { + "location": "rmid" + } + } + } + job = Job(**j) + assert isinstance(job.specifications, TilingSpecifications) + + def test_validation_tui(self): + j = self.j_base.copy() + j["type"] = "TouchUpImport" + j["specifications"] = { + "inputs": { + "modelingReference": "mfid", + "touchUpData": "tud" + }, + "outputs": { + "importInfo": "bkt:tui.json" + } + } + job = Job(**j) + assert isinstance(job.specifications, TouchUpImportSpecifications) + + def test_validation_tue(self): + j = self.j_base.copy() + j["type"] = "TouchUpExport" + j["specifications"] = { + "inputs": { + "modelingReference": "mfid" + }, + "outputs": { + "touchUpData": "data" + } + } + job = Job(**j) + assert isinstance(job.specifications, TouchUpExportSpecifications) + + def test_validation_wc(self): + j = self.j_base.copy() + j["type"] = "WaterConstraints" + j["specifications"] = { + "inputs": { + "modelingReference": "mfid", + "scene": "sid" + }, + "outputs": { + "constraints": "bkt:constraints" + } + } + job = Job(**j) + assert isinstance(job.specifications, WaterConstraintsSpecifications) + + def test_validation_clearance(self): + j = self.j_base.copy() + j["type"] = "ClearanceCalculation" + j["specifications"] = { + "inputs": { + "model3d": "mfid", + "clearanceFootprint": "sid" + }, + "outputs": { + "ovfPoints": "rdId" + } + } + job = Job(**j) + assert isinstance(job.specifications, ClearanceSpecifications) + + def test_validation_unsupported_job_type_raises(self): + j = self.j_base.copy() + j["specifications"] = {"inputs": {}, "outputs": {}} + unsupported = "UnsupportedJobType" + j["type"] = unsupported + with pytest.raises(Exception) as exc_info: + Job.set_specification_validation_model.__func__( + Job, j["specifications"], MagicMock(data={"type": unsupported}) + ) + assert "Unsupported job type" in str(exc_info.value) + diff --git a/python_sdk/tests/test_onprem_jobqueue.py b/python_sdk/tests/test_onprem_jobqueue.py new file mode 100644 index 00000000..bbf8b5c5 --- /dev/null +++ b/python_sdk/tests/test_onprem_jobqueue.py @@ -0,0 +1,308 @@ +import datetime +import shutil +import pytest +import os + +from pydantic import BaseModel, Field + +from reality_capture.common.job import JobState +from reality_capture.on_premise.job import JobPriority, JobFilters +from reality_capture.on_premise.job_manager import JobManager +from reality_capture.on_premise.result import ManagerErrorCode +from reality_capture.specifications.calibration import CalibrationSpecifications, CalibrationInputs, CalibrationOutputs + + +class FakeSpecs(BaseModel): + fake: str = Field(description="Fake field") + + +class TestOnPremJobQueue: + @pytest.fixture(autouse=True) + def tmp_folder(self, tmp_path): + self.tmp_dir = str(tmp_path) + yield + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def test_create_get_cancel(self): + jm = JobManager(self.tmp_dir + "/jq") + ci = CalibrationInputs( + scene="path/to/scene", + ) + co = CalibrationOutputs( + scene="path/to/calibrated_scene", + ) + specs = CalibrationSpecifications(inputs=ci, outputs=co) + job_result = jm.submit_job(specs, self.tmp_dir + "work", workspace=self.tmp_dir + "ws") + assert not job_result.is_error() + assert job_result.value is not None + job = job_result.value + + get_job_result = jm.get_job(job.name) + assert not get_job_result.is_error() + assert get_job_result.value is not None + assert get_job_result.value == job + + set_prio_result = jm.set_job_priority(job.name, JobPriority.URGENT) + assert not set_prio_result.is_error() + assert set_prio_result.value is not None + assert set_prio_result.value.priority == JobPriority.URGENT + + get_job_result = jm.get_job(job.name) + assert not get_job_result.is_error() + assert get_job_result.value is not None + assert get_job_result.value == set_prio_result.value + + get_progress = jm.get_job_progress(job.name) + assert not get_progress.is_error() + assert get_progress.value is not None + assert get_progress.value.state == JobState.QUEUED + assert get_progress.value.percentage == 0 + assert len(get_progress.value.milestones) == 0 + + cancel_result = jm.cancel_job(job.name) + assert not cancel_result.is_error() + assert cancel_result.value is not None + assert cancel_result.value.state == JobState.CANCELLED + + cancel_result = jm.cancel_job(job.name) + assert cancel_result.is_error() + assert cancel_result.error == ManagerErrorCode.JOB_NOT_CANCELLABLE + + cancel_result = jm.cancel_job("wrong_name") + assert cancel_result.is_error() + assert cancel_result.error == ManagerErrorCode.JOB_NOT_FOUND + + def test_with(self): + with JobManager(self.tmp_dir + "/jq") as jm: + res = jm.get_job("wrong_job") + assert res.is_error() + assert res.error == ManagerErrorCode.JOB_NOT_FOUND + + @pytest.fixture + def existing_db(self): + # Copy data from DB_Jobs to tmp dir + current_dir = os.path.dirname(os.path.abspath(__file__)) + source_db_path = os.path.join(current_dir, "data", "DB_Jobs") + target_db_path = os.path.join(self.tmp_dir, "jq") + shutil.copytree(source_db_path, target_db_path) + + def test_summary(self, existing_db): + jm = JobManager(self.tmp_dir + "/jq") + + res = jm.get_summary() + assert not res.is_error() + assert res.value is not None + summary = res.value + assert summary.jobs_queued == 1 + assert len(summary.jobs_active) == 1 + assert summary.jobs_failed == 1 + assert summary.jobs_cancelled == 1 + assert summary.jobs_success == 2 + aj = summary.jobs_active[0] + assert aj.job_name == "job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78" + assert aj.running_tasks == 1 + assert aj.ready_tasks == 1 + + def test_progress(self, existing_db): + jm = JobManager(self.tmp_dir + "/jq") + + res = jm.get_job_progress("job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78") + assert not res.is_error() + assert res.value is not None + progress = res.value + assert progress.state == JobState.ACTIVE + assert progress.percentage == 25.009488 + assert len(progress.milestones) == 2 + + res = jm.get_job_progress("wrong_job") + assert res.is_error() + assert res.error == ManagerErrorCode.JOB_NOT_FOUND + + def test_get_jobs(self, existing_db): + jm = JobManager(self.tmp_dir + "/jq") + jf = JobFilters(includeState=[]) + res = jm.get_jobs(jf) + assert not res.is_error() + assert res.value is not None + jp = res.value + assert jp.next_continuation_token is None + assert len(jp.jobs) == 0 + + jf = JobFilters(limit=2) + res = jm.get_jobs(jf) + assert not res.is_error() + assert res.value is not None + jp = res.value + assert jp.next_continuation_token is not None + assert len(jp.jobs) == 2 + names = [j.name for j in jp.jobs] + + jf = JobFilters(limit=2, continuationToken=jp.next_continuation_token) + res = jm.get_jobs(jf) + assert not res.is_error() + assert res.value is not None + jp = res.value + assert jp.next_continuation_token is not None + assert len(jp.jobs) == 2 + new_names = [j.name for j in jp.jobs] + assert new_names != names + + jf = JobFilters(includeState=[JobState.QUEUED], limit=2) + res = jm.get_jobs(jf) + assert not res.is_error() + assert res.value is not None + jp = res.value + assert jp.next_continuation_token is None + assert len(jp.jobs) == 1 + + sub_time_start = datetime.datetime(2026, 6, 26, 13, 5, 0) + sub_time_end = datetime.datetime(2026, 6, 26, 13, 10, 0) + jf = JobFilters(createdDateTimeRange=(sub_time_start, sub_time_end), limit=50) + res = jm.get_jobs(jf) + assert not res.is_error() + assert res.value is not None + jp = res.value + assert jp.next_continuation_token is None + assert len(jp.jobs) == 1 + assert jp.jobs[0].name == "job_6448e60c-a22a-44f7-819e-1a71068feebd" + + def test_submit_unsupported_job_specs(self): + jm = JobManager(self.tmp_dir + "/jq") + ci = CalibrationInputs( + scene="path/to/scene", + ) + co = CalibrationOutputs( + scene="path/to/calibrated_scene", + ) + specs = CalibrationSpecifications(inputs=ci, outputs=co) + job_result = jm.submit_job(specs, self.tmp_dir + "work") + assert not job_result.is_error() + assert job_result.value is not None + + def test_change_priority_of_nonexistent_job(self, existing_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.set_job_priority("nonexistent_job", JobPriority.HIGH) + assert res.is_error() + assert res.error == ManagerErrorCode.JOB_NOT_FOUND + + @pytest.fixture + def corrupted_db(self): + # Copy data from DB_Jobs to tmp dir + current_dir = os.path.dirname(os.path.abspath(__file__)) + source_db_path = os.path.join(current_dir, "data", "DB_Jobs_Corrupted") + target_db_path = os.path.join(self.tmp_dir, "jq") + shutil.copytree(source_db_path, target_db_path) + + def test_corrupted_json_specs(self, corrupted_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.get_job("job_8aaaffdb-e1a2-451e-97de-9fb2cd5ef8b6") + assert res.is_error() + assert res.error == ManagerErrorCode.CORRUPTED_SPECIFICATIONS + + def test_missing_specs(self, corrupted_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.get_job("job_6448e60c-a22a-44f7-819e-1a71068feebd") + assert res.is_error() + assert res.error == ManagerErrorCode.MISSING_SPECIFICATIONS + + def test_invalid_job_type(self, corrupted_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.get_job("job_12083864-05aa-4a5a-b183-6242fec8bfa5") + assert res.is_error() + assert res.error == ManagerErrorCode.INVALID_JOB_TYPE_IN_DB + + def test_missing_job_type(self, corrupted_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.get_job("job_dec6d7f0-aa2a-4d96-9f26-2f1fa7faa5d5") + assert res.is_error() + assert res.error == ManagerErrorCode.INVALID_JOB_TYPE_IN_DB + + def test_invalid_spec(self, corrupted_db): + jm = JobManager(self.tmp_dir + "/jq") + res = jm.get_job("job_faa0be51-3f34-452b-a4ea-885ebe9015b3") + assert res.is_error() + assert res.error == ManagerErrorCode.CORRUPTED_SPECIFICATIONS + + def test_invalid_continuation_token(self, existing_db): + jm = JobManager(self.tmp_dir + "/jq") + jf = JobFilters(limit=2, continuationToken="invalid_token") + res = jm.get_jobs(jf) + assert res.is_error() + assert res.error == ManagerErrorCode.INVALID_CONTINUATION_TOKEN + + def test_empty_working_dir(self): + jm = JobManager(self.tmp_dir + "/jq") + ci = CalibrationInputs( + scene="path/to/scene", + ) + co = CalibrationOutputs( + scene="path/to/calibrated_scene", + ) + specs = CalibrationSpecifications(inputs=ci, outputs=co) + job_result = jm.submit_job(specs, "") + assert job_result.is_error() + assert job_result.error == ManagerErrorCode.EMPTY_SHARED_WORKING_DIRECTORY + + def test_db_is_locked(self): + jm = JobManager(self.tmp_dir + "/jq") + jm2 = JobManager(self.tmp_dir + "/jq") + ci = CalibrationInputs( + scene="path/to/scene", + ) + co = CalibrationOutputs( + scene="path/to/calibrated_scene", + ) + specs = CalibrationSpecifications(inputs=ci, outputs=co) + + db_path = os.path.join(self.tmp_dir, "jq", "JobQueue.db") + fd = jm2._acquire_lock(db_path, 1) + assert fd is not None + try: + jm._timeout_lock_s = 2 # For speed’s sake + res = jm.submit_job(specs, self.tmp_dir + "/work") + assert res.is_error() + assert res.error == ManagerErrorCode.DB_BUSY + + res = jm.cancel_job("job") + assert res.is_error() + assert res.error == ManagerErrorCode.DB_BUSY + + res = jm.set_job_priority("job", JobPriority.NORMAL) + assert res.is_error() + assert res.error == ManagerErrorCode.DB_BUSY + finally: + jm2._release_lock(fd, db_path) + + def test_db_is_read_only(self, existing_db): + jq_dir = self.tmp_dir + "/jq" + for root, dirs, files in os.walk(jq_dir): + for file in files: + os.chmod(os.path.join(root, file), 0o444) + + jm = JobManager(jq_dir) + ci = CalibrationInputs( + scene="path/to/scene", + ) + co = CalibrationOutputs( + scene="path/to/calibrated_scene", + ) + specs = CalibrationSpecifications(inputs=ci, outputs=co) + + res = jm.submit_job(specs, shared_working_directory="yo") + assert res.is_error() + assert res.error == ManagerErrorCode.SQLITE_ERROR + + res = jm.set_job_priority("job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78", JobPriority.URGENT) + assert res.is_error() + assert res.error == ManagerErrorCode.SQLITE_ERROR + + res = jm.cancel_job("job_d4fc6fa0-8c99-4dff-a7cd-4deb930c4d78") + assert res.is_error() + assert res.error == ManagerErrorCode.SQLITE_ERROR + + def test_fake_specs(self, existing_db): + fs = FakeSpecs(fake="fake") + jm = JobManager(self.tmp_dir + "/jq") + res = jm.submit_job(fs, self.tmp_dir + "/work") + assert res.is_error() + assert res.error == ManagerErrorCode.UNSUPPORTED_SPECIFICATIONS