The software suite for developing and deploying your machine vision applications

def _prepare_devices_configurations(self, devices_configurations: list[DeviceConfiguration]) -> None: 
    super()._prepare_devices_configurations(devices_configurations)
    log.info("Override of _prepare_devices_configurations method (create custom datamatrix reader)") 

    if len([conf for conf in devices_configurations if conf.device_type == DeviceType.code_reader]) == 0 :
        # Automatically create a datamatrix reader device for it to be accessible to operators from PAP
        scanner_conf = DeviceConfiguration(
        label="Datamatrix reader (custom)", 
        device_type=DeviceType.code_reader, 
        device_model="custom", 
        config=TypedConfig.from_dict({
            "id_vendor": {
                "type": "str", 
                "value": "1234"
            }, 
            "id_product": {
                "type": "str", 
                "value": "5678"
            }, 
            "timeout": {
                "type": "float", 
                "value": 1
            }, 
        }),
        )
    devices_configurations.append(scanner_conf) 

    log.info("Created scanner automatically")

class InstancesCounter(DetectionModel):
    """ Adapted version of a detection model to only count instances, for very quick applications with high number of instances """

    def __init__(
        self,
        weights: pathlib.Path,
        preprocess: Optional[PreprocessingT],
        img_size: int = 256,
        force_cpu: bool = False
    ):
        super().__init__(
            weights=weights,
            preprocess=preprocess,
            img_size=img_size,
            force_cpu=force_cpu,
        )

    @torch.inference_mode()
    def infer_get_count_only(self, images_original: np.ndarray, conf_thresh: float = 0.1, iou_thresh: float = 0.5) -> List[int]:
        # If batch dimension not present
        if len(images_original.shape) == 3:
            images_original = np.expand_dims(images_original, 0)

        images = self.preprocess(images_original)
        preds = self.model(images, augment=False)
        preds = non_max_suppression(preds, conf_thresh, iou_thresh)

        nb_instances: List[int] = [0] * len(preds)  # number of instances for each image
        for i, det in enumerate(preds):  # detections per image
            if det is not None and len(det):
                nb_instances[i] = len(det)

        return nb_instances

def _after_application_apply(self) -> None:
    super()._after_application_apply()
    self.stop_stream()

    # Start socket manager
    if (
        self.get_config_value("sockets_enabled", True) is True
        and self.socket_manager is None
    ):
        self.socket_manager = SocketManager(
            trig=self._command_trig,
            expose_params=self._expose_params,
            get_param=self._get_param,
            request_stream=self._request_stream,
            tcp_ip=self.get_config_value("tcp_ip"),
            tcp_port=self.get_config_value("tcp_port"),
            frame_suffix="\r\n",
        )
        log.info("starting sockets")
        self.socket_manager.start()
    else:
        log.info("sockets not enabled")