Skip to content

API reference

Public lifecycle

Own one verified model graph and serialize access to mutable inference sessions.

Source code in ofiqpy/assessor.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Assessor:
    """Own one verified model graph and serialize access to mutable inference sessions."""

    def __init__(self, config: OFIQConfig | None = None):
        self.config = config or OFIQConfig()
        self.pipeline = OFIQPipeline(self.config)
        self.measures = Measures(self.config)
        self._lock = threading.RLock()
        self.pipeline.preload()
        self.measures.preload()

    def assess(self, image: str | os.PathLike[str] | np.ndarray) -> AssessmentResult:
        if isinstance(image, np.ndarray):
            bgr = image
        else:
            bgr = cv2.imread(os.fspath(image))
            if bgr is None:
                return AssessmentResult.failed(FailureDetail(FailureCode.IMAGE_READ_ERROR, f"could not read image: {image}"))
        if bgr.dtype != np.uint8 or bgr.ndim != 3 or bgr.shape[2] != 3 or bgr.shape[0] == 0 or bgr.shape[1] == 0:
            return AssessmentResult.failed(
                FailureDetail(
                    FailureCode.INVALID_IMAGE,
                    f"expected a non-empty BGR uint8 array with shape (height, width, 3); got {bgr.dtype} {bgr.shape}",
                )
            )

        with self._lock:
            try:
                session = self.pipeline.process(bgr)
            except Exception as exc:
                return AssessmentResult.failed(FailureDetail(FailureCode.PIPELINE_ERROR, f"{type(exc).__name__}: {exc}"))
            if session.bbox is None:
                return AssessmentResult.failed(FailureDetail(FailureCode.NO_FACE, "no face detected"))
            return self.measures.compute_typed(session)

Assess one image and retain typed image/component failure states.

Source code in ofiqpy/__init__.py 
25
26
27
def assess_typed(image: "str | object") -> AssessmentResult:
    """Assess one image and retain typed image/component failure states."""
    return _lazy().assess(image)

Assess one image and return {component_name: (raw, scalar)}.

Parameters:

Name Type Description Default
image 'str | object'

path to an image file (str or PathLike), or a BGR uint8 numpy array of shape (H, W, 3).

 required

Returns:

Type Description
dict

Mapping of OFIQ component name to (raw native value, 0-100 scalar). Empty
dict

only if no face is detected.

Raises:

Type Description
FileNotFoundError

The supplied path cannot be decoded as an image.
ValueError

The supplied array is not a non-empty BGR uint8 image.
RuntimeError

Canonical preprocessing fails before component assessment.
Source code in ofiqpy/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def assess(image: "str | object") -> dict:
    """Assess one image and return ``{component_name: (raw, scalar)}``.

    Args:
        image: path to an image file (str or PathLike), or a BGR uint8 numpy
            array of shape (H, W, 3).

    Returns:
        Mapping of OFIQ component name to ``(raw native value, 0-100 scalar)``. Empty
        only if no face is detected.

    Raises:
        FileNotFoundError: The supplied path cannot be decoded as an image.
        ValueError: The supplied array is not a non-empty BGR uint8 image.
        RuntimeError: Canonical preprocessing fails before component assessment.
    """
    result = assess_typed(image)
    if result.failure is not None:
        if result.failure.code is FailureCode.NO_FACE:
            return {}
        if result.failure.code is FailureCode.IMAGE_READ_ERROR:
            raise FileNotFoundError(result.failure.message)
        if result.failure.code is FailureCode.INVALID_IMAGE:
            raise ValueError(result.failure.message)
        raise RuntimeError(f"{result.failure.code.value}: {result.failure.message}")
    return result.as_legacy_dict()

Typed results

Source code in ofiqpy/results.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True)
class AssessmentResult:
    components: Mapping[str, ComponentResult]
    status: AssessmentStatus
    failure: FailureDetail | None = None

    def __post_init__(self) -> None:
        supplied = set(self.components)
        expected = set(OFIQ_COMPONENTS)
        if supplied != expected:
            missing = sorted(expected - supplied)
            extra = sorted(supplied - expected)
            raise ValueError(f"assessment component set mismatch; missing={missing}, extra={extra}")
        ordered = {name: self.components[name] for name in OFIQ_COMPONENTS}
        successful = sum(result.status is ComponentStatus.SUCCESS for result in ordered.values())
        if successful == len(ordered):
            expected_status = AssessmentStatus.SUCCESS
        elif successful:
            expected_status = AssessmentStatus.PARTIAL
        else:
            expected_status = AssessmentStatus.FAILURE_TO_ASSESS
        if self.status is not expected_status:
            raise ValueError(f"assessment status {self.status.value} does not match component results")
        if self.failure is not None and self.status is not AssessmentStatus.FAILURE_TO_ASSESS:
            raise ValueError("an aggregate failure is valid only when the assessment failed")
        object.__setattr__(self, "components", MappingProxyType(ordered))

    @classmethod
    def from_components(cls, components: Mapping[str, ComponentResult]) -> "AssessmentResult":
        supplied = set(components)
        expected = set(OFIQ_COMPONENTS)
        if supplied != expected:
            missing = sorted(expected - supplied)
            extra = sorted(supplied - expected)
            raise ValueError(f"assessment component set mismatch; missing={missing}, extra={extra}")
        ordered = {name: components[name] for name in OFIQ_COMPONENTS}
        successful = sum(result.status is ComponentStatus.SUCCESS for result in ordered.values())
        if successful == len(ordered):
            status = AssessmentStatus.SUCCESS
        elif successful:
            status = AssessmentStatus.PARTIAL
        else:
            status = AssessmentStatus.FAILURE_TO_ASSESS
        return cls(components=ordered, status=status)

    @classmethod
    def failed(cls, failure: FailureDetail) -> "AssessmentResult":
        components = {name: ComponentResult.failed(failure) for name in OFIQ_COMPONENTS}
        return cls(components=components, status=AssessmentStatus.FAILURE_TO_ASSESS, failure=failure)

    def as_legacy_dict(self) -> dict[str, tuple[float, float]]:
        return {name: (result.raw, result.scalar) for name, result in self.components.items()}
Source code in ofiqpy/results.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@dataclass(frozen=True)
class ComponentResult:
    raw: float
    scalar: float
    status: ComponentStatus
    failure: FailureDetail | None = None

    def __post_init__(self) -> None:
        if self.status is ComponentStatus.SUCCESS:
            if self.failure is not None:
                raise ValueError("a successful component cannot carry a failure detail")
            if not math.isfinite(self.raw) or not math.isfinite(self.scalar):
                raise ValueError("successful component values must be finite")
            if not 0.0 <= self.scalar <= 100.0:
                raise ValueError(f"successful component scalar is outside [0, 100]: {self.scalar}")
        else:
            if self.failure is None:
                raise ValueError("a failed component requires a failure detail")
            if self.raw != 0.0 or self.scalar != -1.0:
                raise ValueError("a failed component must use the canonical (0, -1) sentinel")

    @classmethod
    def success(cls, raw: float, scalar: float) -> "ComponentResult":
        return cls(float(raw), float(scalar), ComponentStatus.SUCCESS)

    @classmethod
    def failed(cls, failure: FailureDetail) -> "ComponentResult":
        return cls(0.0, -1.0, ComponentStatus.FAILURE_TO_ASSESS, failure)

Bases: str, Enum

Source code in ofiqpy/results.py 
48
49
50
51
class AssessmentStatus(str, Enum):
    SUCCESS = "success"
    PARTIAL = "partial"
    FAILURE_TO_ASSESS = "failure_to_assess"

Bases: str, Enum

Source code in ofiqpy/results.py 
43
44
45
class ComponentStatus(str, Enum):
    SUCCESS = "success"
    FAILURE_TO_ASSESS = "failure_to_assess"

Bases: str, Enum

Source code in ofiqpy/results.py 
54
55
56
57
58
59
60
class FailureCode(str, Enum):
    NO_FACE = "no_face"
    IMAGE_READ_ERROR = "image_read_error"
    INVALID_IMAGE = "invalid_image"
    PIPELINE_ERROR = "pipeline_error"
    COMPONENT_ERROR = "component_error"
    COMPONENT_UNAVAILABLE = "component_unavailable"

Verified profile and configuration

Source code in ofiqpy/profile.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@dataclass(frozen=True)
class CanonicalProfile:
    profile_id: str
    config_sha256: str
    artifacts: tuple[ArtifactSpec, ...]

    def verify(self, data_root: Path) -> ProfileVerification:
        root = Path(data_root)
        config_path = root / "ofiq_config.jaxn"
        if not config_path.is_file():
            raise ProfileVerificationError(f"canonical OFIQ config is missing: {config_path}")
        actual_config_hash = _sha256(config_path)
        if actual_config_hash != self.config_sha256:
            raise ProfileVerificationError(
                f"OFIQ config hash mismatch for {config_path}: expected {self.config_sha256}, got {actual_config_hash}"
            )

        verified_bytes = 0
        for artifact in self.artifacts:
            path = root / artifact.relative_path
            if not path.is_file():
                raise ProfileVerificationError(f"canonical OFIQ artifact is missing: {path}")
            actual_size = path.stat().st_size
            if actual_size != artifact.size_bytes:
                raise ProfileVerificationError(
                    f"OFIQ artifact size mismatch for {path}: expected {artifact.size_bytes}, got {actual_size}"
                )
            actual_hash = _sha256(path)
            if actual_hash != artifact.sha256:
                raise ProfileVerificationError(
                    f"OFIQ artifact hash mismatch for {path}: expected {artifact.sha256}, got {actual_hash}"
                )
            verified_bytes += actual_size

        return ProfileVerification(
            profile_id=self.profile_id,
            data_root=root,
            config_sha256=actual_config_hash,
            verified_artifacts=len(self.artifacts),
            verified_bytes=verified_bytes,
        )

Verified canonical OFIQ v1.1.0 config and model-path resolver.

Source code in ofiqpy/config.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class OFIQConfig:
    """Verified canonical OFIQ v1.1.0 config and model-path resolver."""

    def __init__(self, path: Path | None = None, data_root: Path | None = None):
        if data_root is None:
            data_root = Path(path).parent if path is not None else default_data_root()
        self.data_root = Path(data_root)
        canonical_path = self.data_root / "ofiq_config.jaxn"
        self.path = Path(path) if path is not None else canonical_path
        if self.path != canonical_path:
            raise ValueError(
                f"the canonical profile requires config and models from one data root: expected {canonical_path}, got {self.path}"
            )
        self.profile: CanonicalProfile = CANONICAL_PROFILE
        self.verification: ProfileVerification = self.profile.verify(self.data_root)
        self.cfg = load_config(self.path)
        self.params = self.cfg["params"]

    def resolve(self, rel_path: str) -> Path:
        p = self.data_root / rel_path
        if not p.exists():
            raise FileNotFoundError(f"OFIQ model not found: {p}")
        return p

    def measure(self, name: str) -> dict:
        return self.params["measures"].get(name, {})

    def sigmoid_params(self, measure: str) -> dict:
        """Return canonical JAXN values for inspection; runtime overrides are unsupported."""
        return self.measure(measure).get("Sigmoid", {})

    def detector(self) -> dict:
        return self.params["detector"]["ssd"]

    def landmarks_model(self) -> Path:
        return self.resolve(self.params["landmarks"]["ADNet"]["model_path"])

sigmoid_params(measure)

Return canonical JAXN values for inspection; runtime overrides are unsupported.

Source code in ofiqpy/config.py 
94
95
96
def sigmoid_params(self, measure: str) -> dict:
    """Return canonical JAXN values for inspection; runtime overrides are unsupported."""
    return self.measure(measure).get("Sigmoid", {})

Advanced pipeline access

OFIQPipeline and Measures expose internal preprocessing and execution stages. They do not provide Assessor's lock, array validation, or typed image-level failure boundary.

Source code in ofiqpy/pipeline.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class OFIQPipeline:
    def __init__(self, cfg: OFIQConfig | None = None, enable_pose=True, enable_parsing=True, enable_occlusion=True):
        # The pip wheel's optimized float-resize path differs from OFIQ's conan
        # OpenCV 4.5.5 build. The generic path is bit-identical on the reviewed tensors.
        cv2.setUseOptimized(False)
        self.cfg = cfg or OFIQConfig()
        d = self.cfg.detector()
        self.detector = SSDDetector(
            self.cfg.resolve(d["model_path"]),
            self.cfg.resolve(d["prototxt_path"]),
            d["confidence_thr"],
            d["min_rel_face_size"],
            d["padding"],
        )
        self.landmarker = ADNetLandmarker(self.cfg.landmarks_model())
        self._pose = self._parser = self._occ = None
        self.enable_pose = enable_pose
        self.enable_parsing = enable_parsing
        self.enable_occlusion = enable_occlusion

    def _get_pose(self):
        if self._pose is None:
            from .pose.tddfa import TDDFAPose

            self._pose = TDDFAPose(self.cfg.resolve(self.cfg.measure("HeadPose")["model_path"]))
        return self._pose

    def _get_parser(self):
        if self._parser is None:
            from .segmentation.parsing import FaceParser

            self._parser = FaceParser(self.cfg.resolve(self.cfg.measure("FaceParsing")["model_path"]))
        return self._parser

    def _get_occ(self):
        if self._occ is None:
            from .segmentation.occlusion import OcclusionSeg

            self._occ = OcclusionSeg(self.cfg.resolve(self.cfg.measure("FaceOcclusionSegmentation")["model_path"]))
        return self._occ

    def preload(self) -> None:
        """Load every enabled canonical preprocessing model before assessment starts."""
        if self.enable_pose:
            self._get_pose()
        if self.enable_parsing:
            self._get_parser()
        if self.enable_occlusion:
            self._get_occ()

    def process(self, image: np.ndarray) -> Session:
        s = Session(image=image)
        faces = self.detector.detect(image)
        s.n_faces = len(faces)
        s.face_areas = [f[2] * f[3] for f in faces]
        if not faces:
            return s
        primary = faces[0]
        s.bbox = np.array([primary[0], primary[1], primary[0] + primary[2], primary[1] + primary[3]])
        if self.enable_pose:
            yaw, pitch, roll = self._get_pose().estimate(image, primary)
            s.yaw, s.pitch, s.roll = yaw, pitch, roll
        s.landmarks = self.landmarker.extract(image, primary)
        s.aligned_face, s.aligned_landmarks, s.affine = align(image, s.landmarks)
        if self.enable_parsing:
            s.parsing = self._get_parser().parse(s.aligned_face)
        if self.enable_occlusion:
            s.occlusion_mask = self._get_occ().segment(s.aligned_face)
        s.landmarked_region = landmarked_region(s.aligned_landmarks)
        return s

preload()

Load every enabled canonical preprocessing model before assessment starts.

Source code in ofiqpy/pipeline.py 
60
61
62
63
64
65
66
67
def preload(self) -> None:
    """Load every enabled canonical preprocessing model before assessment starts."""
    if self.enable_pose:
        self._get_pose()
    if self.enable_parsing:
        self._get_parser()
    if self.enable_occlusion:
        self._get_occ()
Source code in ofiqpy/session.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@dataclass
class Session:
    image: np.ndarray  # original BGR uint8 (H, W, 3)

    # detection
    bbox: np.ndarray | None = None  # (4,) [x1, y1, x2, y2] in original px
    n_faces: int = 0
    face_areas: list[float] = field(default_factory=list)  # descending

    # landmarks (ADNet 98-pt), original-image pixel coords
    landmarks: np.ndarray | None = None  # (98, 2)

    # alignment (616x616 canonical) + transformed landmarks
    aligned_face: np.ndarray | None = None  # (616, 616, 3) BGR uint8
    aligned_landmarks: np.ndarray | None = None  # (98, 2) in aligned space
    affine: np.ndarray | None = None  # (2, 3) transform used

    # pose (degrees)
    yaw: float | None = None
    pitch: float | None = None
    roll: float | None = None

    # segmentation
    parsing: np.ndarray | None = None  # aligned parsing label map
    occlusion_mask: np.ndarray | None = None  # aligned binary occlusion mask

    # derived masks
    landmarked_region: np.ndarray | None = None  # aligned face-region mask

    def has(self, *attrs: str) -> bool:
        return all(getattr(self, a) is not None for a in attrs)
Source code in ofiqpy/measures/core.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
class Measures:
    def __init__(self, cfg):
        self.cfg = cfg
        self._ssim = None
        self._magface = None
        self._rtree = None
        self._num_trees = None
        self._enet1 = self._enet2 = self._boost = None

    def _load_ml_gz(self, path, loader):
        import gzip
        import tempfile

        data = gzip.decompress(Path(path).read_bytes())
        suffix = ".xml" if str(path).endswith(".xml.gz") else ".yml"
        with tempfile.NamedTemporaryFile("wb", suffix=suffix, delete=False) as f:
            f.write(data)
            tmp = f.name
        try:
            return loader(tmp)
        finally:
            Path(tmp).unlink(missing_ok=True)

    def _rtree_model(self):
        if self._rtree is None:
            p = self.cfg.resolve(self.cfg.measure("Sharpness")["model_path"])
            self._rtree = self._load_ml_gz(p, cv2.ml.RTrees_load)
            self._num_trees = int(self._rtree.getTermCriteria()[1])
        return self._rtree, self._num_trees

    def _expression_models(self):
        if self._enet1 is None:
            m = self.cfg.measure("ExpressionNeutrality")
            self._enet1 = ort.InferenceSession(str(self.cfg.resolve(m["cnn1_model_path"])), providers=["CPUExecutionProvider"])
            self._enet2 = ort.InferenceSession(str(self.cfg.resolve(m["cnn2_model_path"])), providers=["CPUExecutionProvider"])
            self._boost = self._load_ml_gz(self.cfg.resolve(m["adaboost_model_path"]), cv2.ml.Boost_load)
        return self._enet1, self._enet2, self._boost

    def _ssim_sess(self):
        if self._ssim is None:
            p = self.cfg.resolve(self.cfg.measure("CompressionArtifacts")["model_path"])
            self._ssim = ort.InferenceSession(str(p), providers=["CPUExecutionProvider"])
        return self._ssim

    def _magface_sess(self):
        if self._magface is None:
            p = self.cfg.resolve(self.cfg.measure("UnifiedQualityScore")["model_path"])
            self._magface = ort.InferenceSession(str(p), providers=["CPUExecutionProvider"])
        return self._magface

    def preload(self) -> None:
        """Load every canonical measure model before output creation."""
        self._ssim_sess()
        self._magface_sess()
        self._rtree_model()
        self._expression_models()

    # --- C03 LuminanceMean (hardcoded double-sigmoid mapping) ---
    def luminance_mean(self, s):
        L = luminance(s.aligned_face)
        hist = cv2.calcHist([L], [0], s.landmarked_region, [256], [0, 256]).flatten()
        hist = hist / hist.sum()
        mean = float(hist @ np.arange(256)) / 255.0
        scalar = _round_half_away(100.0 * _sigmoid(mean, 0.2, 0.05) * (1.0 - _sigmoid(mean, 0.8, 0.05)))
        return MeasureValue.success("LuminanceMean", mean, max(0.0, min(100.0, scalar)))

    # --- C20 HeadSize ---
    def head_size(self, s):
        T = tmetric(s.landmarks)  # ORIGINAL landmarks
        raw = T / s.image.shape[0]  # original height
        cs = abs(raw - 0.45)
        scalar = scalar_conversion(cs, h=200, a=1.0, s=-1.0, x0=0.0, w=0.05, round=True)
        return MeasureValue.success("HeadSize", raw, scalar)

    # --- C17 NoHeadCoverings (custom piecewise mapping) ---
    def no_head_coverings(self, s):
        M = s.parsing  # 400x400
        crop = M[0 : 400 - 204, :]  # top 196 rows
        n = int((crop == CLOTH).sum() + (crop == HAT).sum())
        raw = n / (400 * 196)
        T0, T1, w, x0 = 0.0, 0.95, 0.1, 0.02
        if raw <= T0:
            scalar = 100.0
        elif raw >= T1:
            scalar = 0.0
        else:
            sv = _sigmoid(raw, x0, w)
            s0 = _sigmoid(T0, x0, w)
            s1 = _sigmoid(T1, x0, w)
            scalar = _round_half_away(100.0 * (s1 - sv) / (s1 - s0))
        return MeasureValue.success("NoHeadCoverings", raw, max(0.0, min(100.0, scalar)))

    # --- C09 CompressionArtifacts (reuses OFIQ ssim_248 ONNX) ---
    def compression(self, s):
        crop = s.aligned_face[184 : 616 - 184, 184 : 616 - 184]  # 248x248
        rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB).astype(np.float32)
        transformed = cv2.subtract(rgb, (123.7, 116.3, 103.5, 0.0))
        transformed = cv2.divide(transformed, (58.4, 57.1, 57.4, 0.0))
        blob = np.ascontiguousarray(transformed.transpose(2, 0, 1)[None])
        raw = float(self._ssim_sess().run(None, {"input": blob})[0].reshape(-1)[0])
        scalar = scalar_conversion(raw, h=1, a=-0.0278, s=103.0, x0=0.3308, w=0.092, round=True)
        return MeasureValue.success("CompressionArtifacts", raw, scalar)

    # --- UnifiedQualityScore (MagFace magnitude -> sigmoid) ---
    def unified(self, s):
        resized = cv2.resize(s.aligned_face, (192, 192), interpolation=cv2.INTER_LINEAR)
        crop = resized[33 : 192 - 47, 40 : 192 - 40]  # 112x112, BGR
        conv = crop.astype(np.float32) / 255.0
        blob = np.transpose(conv, (2, 0, 1))[None]
        raw = float(self._magface_sess().run(None, {"input": blob})[0].reshape(-1)[0])
        scalar = scalar_conversion(raw, h=100, a=0.0, s=1.0, x0=23.0, w=2.6, round=True)
        return MeasureValue.success("UnifiedQualityScore", raw, scalar)

    # --- HeadPose Yaw/Pitch/Roll (with the OFIQ slot swap) ---
    @staticmethod
    def _cos2(angle_deg):
        c = max(0.0, math.cos(math.radians(angle_deg)))
        return _round_half_away(100.0 * c * c)

    def head_pose(self, s):
        yaw, pitch, roll = s.yaw, s.pitch, s.roll  # geometric
        names = ("HeadPoseYaw", "HeadPosePitch", "HeadPoseRoll")
        if yaw is None or pitch is None or roll is None:
            return tuple(MeasureValue.unavailable(name) for name in names)
        # slot swap: HeadPoseYaw<-pitch, HeadPosePitch<-yaw, HeadPoseRoll<-roll
        # native value = the (swapped) angle in degrees
        return (
            MeasureValue.success("HeadPoseYaw", pitch, self._cos2(pitch)),
            MeasureValue.success("HeadPosePitch", yaw, self._cos2(yaw)),
            MeasureValue.success("HeadPoseRoll", roll, self._cos2(roll)),
        )

    @staticmethod
    def _execute_group(
        out: dict[str, ComponentResult],
        expected_names: tuple[str, ...],
        producer: Callable[[], MeasureValue | tuple[MeasureValue, ...]],
    ) -> None:
        """Run and validate one native OFIQ component group."""
        try:
            produced = producer()
            values = (produced,) if isinstance(produced, MeasureValue) else tuple(produced)
            if not all(isinstance(value, MeasureValue) for value in values):
                raise TypeError("measure producer returned a value outside the MeasureValue contract")
            actual_names = tuple(value.name for value in values)
            if actual_names != expected_names:
                raise ValueError(f"measure producer names {actual_names} do not match expected names {expected_names}")
            for value in values:
                if value.raw is not None and value.scalar is not None:
                    out[value.name] = ComponentResult.success(value.raw, value.scalar)
                else:
                    out[value.name] = ComponentResult.failed(
                        FailureDetail(FailureCode.COMPONENT_UNAVAILABLE, f"{value.name} returned FailureToAssess")
                    )
        except Exception as exc:
            detail = FailureDetail(FailureCode.COMPONENT_ERROR, f"{type(exc).__name__}: {exc}")
            for name in expected_names:
                out[name] = ComponentResult.failed(detail)

    def compute_typed(self, s) -> AssessmentResult:
        """Compute all components, isolating a failure to its owning component group."""
        from . import geometry as G
        from . import models as M
        from . import pixel as P

        out: dict[str, ComponentResult] = {}
        groups: list[tuple[tuple[str, ...], Callable[[], MeasureValue | tuple[MeasureValue, ...]]]] = [
            (("LuminanceMean",), lambda: self.luminance_mean(s)),
            (("HeadSize",), lambda: self.head_size(s)),
            (("NoHeadCoverings",), lambda: self.no_head_coverings(s)),
            (("CompressionArtifacts",), lambda: self.compression(s)),
            (("UnifiedQualityScore",), lambda: self.unified(s)),
            (("InterEyeDistance",), lambda: G.inter_eye_distance(s)),
            (("SingleFacePresent",), lambda: G.single_face_present(s)),
            (("EyesOpen",), lambda: G.eyes_open(s)),
            (("MouthClosed",), lambda: G.mouth_closed(s)),
        ]
        crop_names = (
            "LeftwardCropOfTheFaceImage",
            "RightwardCropOfTheFaceImage",
            "MarginAboveOfTheFaceImage",
            "MarginBelowOfTheFaceImage",
        )
        groups.extend(
            [
                (crop_names, lambda: G.crop_of_face(s)),
                (("BackgroundUniformity",), lambda: P.background_uniformity(s)),
                (("IlluminationUniformity",), lambda: P.illumination_uniformity(s)),
                (("LuminanceVariance",), lambda: P.luminance_variance(s)),
                (("UnderExposurePrevention",), lambda: P.under_exposure(s)),
                (("OverExposurePrevention",), lambda: P.over_exposure(s)),
                (("DynamicRange",), lambda: P.dynamic_range(s)),
                (("NaturalColour",), lambda: P.natural_colour(s)),
                (("EyesVisible",), lambda: M.eyes_visible(s)),
                (("MouthOcclusionPrevention",), lambda: M.mouth_occlusion(s)),
                (("FaceOcclusionPrevention",), lambda: M.face_occlusion(s)),
            ]
        )

        def sharpness() -> MeasureValue:
            rtree, nt = self._rtree_model()
            return M.sharpness(s, rtree, nt)

        def expression() -> MeasureValue:
            e1, e2, bo = self._expression_models()
            return M.expression_neutrality(s, e1, e2, bo)

        pose_names = ("HeadPoseYaw", "HeadPosePitch", "HeadPoseRoll")
        groups.extend(
            [
                (("Sharpness",), sharpness),
                (("ExpressionNeutrality",), expression),
                (pose_names, lambda: self.head_pose(s)),
            ]
        )

        for expected_names, producer in groups:
            self._execute_group(out, expected_names, producer)

        missing = [name for name in OFIQ_COMPONENTS if name not in out]
        if missing:
            detail = FailureDetail(FailureCode.COMPONENT_UNAVAILABLE, "component did not produce a result")
            for name in missing:
                out[name] = ComponentResult.failed(detail)
        return AssessmentResult.from_components(out)

    def compute(self, s) -> dict:
        """Compatibility mapping of component name to ``(raw, scalar)``."""
        return self.compute_typed(s).as_legacy_dict()

    def compute_scalars(self, s) -> dict:
        """Return the compatibility mapping of component name to scalar."""
        return {k: v[1] for k, v in self.compute(s).items()}

compute(s)

Compatibility mapping of component name to (raw, scalar).

Source code in ofiqpy/measures/core.py 
251
252
253
def compute(self, s) -> dict:
    """Compatibility mapping of component name to ``(raw, scalar)``."""
    return self.compute_typed(s).as_legacy_dict()

compute_scalars(s)

Return the compatibility mapping of component name to scalar.

Source code in ofiqpy/measures/core.py 
255
256
257
def compute_scalars(self, s) -> dict:
    """Return the compatibility mapping of component name to scalar."""
    return {k: v[1] for k, v in self.compute(s).items()}

compute_typed(s)

Compute all components, isolating a failure to its owning component group.

Source code in ofiqpy/measures/core.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def compute_typed(self, s) -> AssessmentResult:
    """Compute all components, isolating a failure to its owning component group."""
    from . import geometry as G
    from . import models as M
    from . import pixel as P

    out: dict[str, ComponentResult] = {}
    groups: list[tuple[tuple[str, ...], Callable[[], MeasureValue | tuple[MeasureValue, ...]]]] = [
        (("LuminanceMean",), lambda: self.luminance_mean(s)),
        (("HeadSize",), lambda: self.head_size(s)),
        (("NoHeadCoverings",), lambda: self.no_head_coverings(s)),
        (("CompressionArtifacts",), lambda: self.compression(s)),
        (("UnifiedQualityScore",), lambda: self.unified(s)),
        (("InterEyeDistance",), lambda: G.inter_eye_distance(s)),
        (("SingleFacePresent",), lambda: G.single_face_present(s)),
        (("EyesOpen",), lambda: G.eyes_open(s)),
        (("MouthClosed",), lambda: G.mouth_closed(s)),
    ]
    crop_names = (
        "LeftwardCropOfTheFaceImage",
        "RightwardCropOfTheFaceImage",
        "MarginAboveOfTheFaceImage",
        "MarginBelowOfTheFaceImage",
    )
    groups.extend(
        [
            (crop_names, lambda: G.crop_of_face(s)),
            (("BackgroundUniformity",), lambda: P.background_uniformity(s)),
            (("IlluminationUniformity",), lambda: P.illumination_uniformity(s)),
            (("LuminanceVariance",), lambda: P.luminance_variance(s)),
            (("UnderExposurePrevention",), lambda: P.under_exposure(s)),
            (("OverExposurePrevention",), lambda: P.over_exposure(s)),
            (("DynamicRange",), lambda: P.dynamic_range(s)),
            (("NaturalColour",), lambda: P.natural_colour(s)),
            (("EyesVisible",), lambda: M.eyes_visible(s)),
            (("MouthOcclusionPrevention",), lambda: M.mouth_occlusion(s)),
            (("FaceOcclusionPrevention",), lambda: M.face_occlusion(s)),
        ]
    )

    def sharpness() -> MeasureValue:
        rtree, nt = self._rtree_model()
        return M.sharpness(s, rtree, nt)

    def expression() -> MeasureValue:
        e1, e2, bo = self._expression_models()
        return M.expression_neutrality(s, e1, e2, bo)

    pose_names = ("HeadPoseYaw", "HeadPosePitch", "HeadPoseRoll")
    groups.extend(
        [
            (("Sharpness",), sharpness),
            (("ExpressionNeutrality",), expression),
            (pose_names, lambda: self.head_pose(s)),
        ]
    )

    for expected_names, producer in groups:
        self._execute_group(out, expected_names, producer)

    missing = [name for name in OFIQ_COMPONENTS if name not in out]
    if missing:
        detail = FailureDetail(FailureCode.COMPONENT_UNAVAILABLE, "component did not produce a result")
        for name in missing:
            out[name] = ComponentResult.failed(detail)
    return AssessmentResult.from_components(out)

preload()

Load every canonical measure model before output creation.

Source code in ofiqpy/measures/core.py 
75
76
77
78
79
80
def preload(self) -> None:
    """Load every canonical measure model before output creation."""
    self._ssim_sess()
    self._magface_sess()
    self._rtree_model()
    self._expression_models()

Output and batch

Source code in ofiqpy/output.py 
32
33
def header() -> str:
    return _encode(header_fields())

results: {component: (raw, scalar)}. Missing components -> FailureToAssess.

Source code in ofiqpy/output.py 
36
37
38
39
40
41
42
43
44
45
def row(filename: str, results: dict | AssessmentResult, time_ms: float) -> str:
    """results: {component: (raw, scalar)}. Missing components -> FailureToAssess."""
    values = results.as_legacy_dict() if isinstance(results, AssessmentResult) else results
    raws, scalars = [], []
    for c in OFIQ_ORDER:
        raw, scalar = values.get(c, FAILURE)
        raws.append(f"{raw:.6f}" if raw is not None else "nan")
        scalars.append(str(int(scalar)) if scalar is not None else "-1")
    parts = [filename] + raws + scalars + [f"{time_ms:.0f}"]
    return _encode(parts)
Source code in ofiqpy/batch.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def run_batch(
    input_path,
    output_csv,
    workers=1,
    resume=False,
    progress: Callable[[BatchProgress], None] | bool | None = None,
    include_per_image_provenance: bool = False,
) -> BatchReport:
    if workers is None or workers < 1:
        raise ValueError("workers must be an integer greater than or equal to one")
    source = Path(input_path)
    images = discover(source)
    progress_callback: Callable[[BatchProgress], None] | None
    if progress is True:
        progress_callback = _console_progress
    elif progress is False:
        progress_callback = None
    else:
        progress_callback = progress
    out = Path(output_csv)
    done = _already_done(out) if resume else set()
    todo = [im for im in images if str(im) not in done]
    skipped = len(images) - len(todo)
    provenance_base = source if source.is_dir() else source.parent
    provenance = build_input_provenance(images, provenance_base, include_records=include_per_image_provenance)

    write_header = not (resume and out.exists())
    mode = "a" if (resume and out.exists()) else "w"
    assessed = successful = partial = failure_to_assess = 0
    started = perf_counter()
    if not todo:
        return BatchReport(
            schema="ofiqpy.batch-report.v1",
            output_csv=out,
            discovered=len(images),
            assessed=0,
            skipped=skipped,
            successful=0,
            partial=0,
            failure_to_assess=0,
            elapsed_seconds=perf_counter() - started,
            provenance=provenance,
        )

    executor = None
    encoded_rows: Iterator[_EncodedAssessment]
    if workers == 1:
        assessor = Assessor()
        encoded_rows = (_assess_with(assessor, str(image)) for image in todo)
    else:
        OFIQConfig()
        executor = ProcessPoolExecutor(
            max_workers=workers,
            initializer=_init_worker,
            mp_context=get_context("spawn"),
        )
        encoded_rows = executor.map(_assess_one, [str(im) for im in todo], chunksize=1)

    try:
        first_result = next(encoded_rows)
        with open(out, mode, encoding="utf-8", newline="") as fh:
            if write_header:
                fh.write(header() + "\n")
            for encoded in chain((first_result,), encoded_rows):
                fh.write(encoded.line + "\n")
                fh.flush()
                assessed += 1
                successful += encoded.status is AssessmentStatus.SUCCESS
                partial += encoded.status is AssessmentStatus.PARTIAL
                failure_to_assess += encoded.status is AssessmentStatus.FAILURE_TO_ASSESS
                if progress_callback is not None:
                    elapsed = perf_counter() - started
                    progress_callback(
                        BatchProgress(
                            processed=assessed,
                            total=len(todo),
                            skipped=skipped,
                            elapsed_seconds=elapsed,
                            images_per_second=assessed / max(elapsed, 1e-9),
                        )
                    )
    finally:
        if executor is not None:
            executor.shutdown(cancel_futures=True)
    return BatchReport(
        schema="ofiqpy.batch-report.v1",
        output_csv=out,
        discovered=len(images),
        assessed=assessed,
        skipped=skipped,
        successful=successful,
        partial=partial,
        failure_to_assess=failure_to_assess,
        elapsed_seconds=perf_counter() - started,
        provenance=provenance,
    )

Conformance

Run live OFIQ and ofiqpy with strict row/component/status cardinality.

Source code in ofiqpy/conformance.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def run_conformance(
    *,
    ofiq_root: Path,
    image_dir: Path,
    expected_count: int,
    scalar_tolerance: float,
    source_root: Path | None = None,
    distribution: Path | None = None,
) -> ConformanceReport:
    """Run live OFIQ and ofiqpy with strict row/component/status cardinality."""
    if expected_count < 1:
        raise ValueError("expected_count must be greater than zero")
    if scalar_tolerance < 0:
        raise ValueError("scalar_tolerance must not be negative")
    ofiq_root = Path(ofiq_root).absolute()
    image_dir = Path(image_dir).absolute()
    source_root = Path(source_root or Path(__file__).resolve().parents[1]).absolute()
    distribution = Path(distribution).absolute() if distribution is not None else None
    images = discover(image_dir)
    if len(images) != expected_count:
        raise ConformanceContractError(f"input cardinality mismatch: expected {expected_count}, discovered {len(images)}")
    expected_identities = {_absolute_identity(image) for image in images}
    if len(expected_identities) != expected_count:
        raise ConformanceContractError("input identities are not unique")

    artifact_bindings = _artifact_bindings(ofiq_root, image_dir, images, source_root, distribution)

    reference_rows = _run_reference(ofiq_root, image_dir)
    reference = _index_reference(reference_rows, ofiq_root)
    port = _run_port(images, ofiq_root / "data")
    if set(reference) != expected_identities:
        missing = sorted(expected_identities - set(reference))
        unexpected = sorted(set(reference) - expected_identities)
        raise ConformanceContractError(f"reference identity mismatch: missing={missing}, unexpected={unexpected}")
    if set(port) != expected_identities:
        missing = sorted(expected_identities - set(port))
        unexpected = sorted(set(port) - expected_identities)
        raise ConformanceContractError(f"port identity mismatch: missing={missing}, unexpected={unexpected}")

    comparisons: dict[str, ComponentComparison] = {}
    raw_places = CANONICAL_RAW_TOLERANCE_PROFILE.decimal_places
    raw_scale = 10**raw_places
    for component in OFIQ_COMPONENTS:
        policy = CANONICAL_RAW_TOLERANCE_PROFILE.components[component]
        tolerance_ticks = _raw_ticks(f"{policy.absolute_tolerance:.{raw_places}f}", raw_places)
        scalar_exact = scalar_within = status_exact = raw_observations = raw_exact = raw_within = 0
        raw_deltas: list[float] = []
        max_scalar_delta = max_raw_delta = 0.0
        for identity in sorted(expected_identities):
            reference_scalar = _number(reference[identity][f"{component}.scalar"], identity, component, "scalar")
            port_result = port[identity][component]
            port_scalar = _number(port_result.scalar, identity, component, "port scalar")
            port_success = port_result.status is ComponentStatus.SUCCESS
            reference_success = reference_scalar != -1.0
            reference_raw_text = reference[identity][component] if reference_success else None
            reference_raw = _number(reference_raw_text, identity, component, "raw") if reference_raw_text is not None else None
            port_raw = _number(port_result.raw, identity, component, "port raw") if port_success else None
            status_exact += port_success == reference_success
            scalar_delta = abs(port_scalar - reference_scalar)
            scalar_exact += scalar_delta == 0.0
            scalar_within += scalar_delta <= scalar_tolerance
            max_scalar_delta = max(max_scalar_delta, scalar_delta)
            if reference_raw is not None and reference_raw_text is not None and port_raw is not None:
                port_reported = f"{port_raw:.{raw_places}f}"
                reference_ticks = _raw_ticks(reference_raw_text, raw_places)
                port_ticks = _raw_ticks(port_reported, raw_places)
                delta_ticks = abs(port_ticks - reference_ticks)
                raw_delta = delta_ticks / raw_scale
                raw_observations += 1
                raw_exact += delta_ticks == 0
                raw_within += delta_ticks <= tolerance_ticks
                raw_deltas.append(raw_delta)
                max_raw_delta = max(max_raw_delta, raw_delta)
        raw_passed = raw_within == raw_observations
        passed = scalar_within == expected_count and status_exact == expected_count and raw_passed
        comparisons[component] = ComponentComparison(
            observations=expected_count,
            scalar_exact=scalar_exact,
            scalar_within_tolerance=scalar_within,
            status_exact=status_exact,
            raw_observations=raw_observations,
            raw_exact_at_6_decimals=raw_exact,
            raw_within_tolerance=raw_within,
            raw_tolerance=policy.absolute_tolerance,
            raw_mean_delta=sum(raw_deltas) / len(raw_deltas) if raw_deltas else 0.0,
            raw_p95_delta=_p95(raw_deltas),
            max_scalar_delta=max_scalar_delta,
            max_raw_delta=max_raw_delta,
            raw_passed=raw_passed,
            passed=passed,
        )

    component_observations = expected_count * len(OFIQ_COMPONENTS)
    raw_observations = sum(item.raw_observations for item in comparisons.values())
    final_artifact_bindings = _artifact_bindings(ofiq_root, image_dir, images, source_root, distribution)
    _assert_stable_artifacts(artifact_bindings, final_artifact_bindings)
    return ConformanceReport(
        passed=all(comparison.passed for comparison in comparisons.values()),
        expected_rows=expected_count,
        reference_rows=len(reference),
        port_rows=len(port),
        scalar_tolerance=scalar_tolerance,
        component_observations=component_observations,
        scalar_exact=sum(item.scalar_exact for item in comparisons.values()),
        scalar_within_tolerance=sum(item.scalar_within_tolerance for item in comparisons.values()),
        status_exact=sum(item.status_exact for item in comparisons.values()),
        raw_observations=raw_observations,
        raw_excluded=component_observations - raw_observations,
        raw_exact_at_6_decimals=sum(item.raw_exact_at_6_decimals for item in comparisons.values()),
        raw_tolerance_profile_id=CANONICAL_RAW_TOLERANCE_PROFILE.profile_id,
        raw_within_tolerance=sum(item.raw_within_tolerance for item in comparisons.values()),
        raw_passed=all(item.raw_passed for item in comparisons.values()),
        components=comparisons,
        bindings={**artifact_bindings, **_runtime_bindings()},
    )