def __init__()

in projects/vision-ai-edge-platform/camera-client/edge_camera.py [0:0]

• 48 lines of code
• 7 McCabe index (conditional complexity)

    def __init__(self, address, device_id, gentl, logger, stdout, protobuf):
        """Initializes the instance with Genicam address, from 0 onwards.

        Args:
          address: str, Genicam device address, from 0 onwards.
          device_id: str, Unique identifier for the camera instance.
          gentl: str, Path to the GenTL producer file.
          logger: logging.Logger, Logger object for recording messages.
          stdout: str, Output mode for messages ('print' for stdout, 'protobuf'
            for protobuf messages).
          protobuf: bool, Whether to use protobuf for output messages.
        """
        self.address = address
        self.device_id = device_id
        self.gentl = gentl
        self.logger = logger
        self.stdout = stdout
        self.printout = self.stdout == "print"
        self.protobuf = protobuf
        self.cameras_list = []
        self.img_acquirer = None
        self.cam_proto = cameras_pb2.Camera()
        self.health_proto = cameras_pb2.CameraHealthCheckResult()
        self.frame_proto = cameras_pb2.CameraFrameResult()
        self.harvester = Harvester(logger=logger)
        self.harvester.add_file(gentl)
        self.harvester.timeout_for_update = 5000
        self.cameras_list = self.scan()
        self.discovery_proto = self._discovery_results(self.cameras_list)

        if self.printout:
            print("Genicam cameras found:")
            for cam in self.cameras_list:
                print(
                    "Address: {} | Make: {} | Model: {}".format(
                        cam["address"], cam["make"], cam["model"]
                    )
                )
        elif "protobuf" in self.stdout:
            self.logger.debug("Printing scan results in protobuf")
            print_without_eol(
                self.discovery_proto.SerializeToString().decode("utf-8")
            )

        if self.address:
            stream_proto = cameras_pb2.Camera.Stream()
            stream_proto.protocol = cameras_pb2.Camera.Stream.PROTOCOL_GENICAM
            stream_proto.address = self.address
            self.cam_proto.streams.append(stream_proto)
            for cam in self.cameras_list:
                if self.address == cam["address"]:
                    self.cam_proto.make = cam["make"]
                    self.cam_proto.model = cam["model"]
            self.logger.debug(
                "Opening capture connection to: {}".format(self.address)
            )
            self.img_acquirer = self.harvester.create(
                search_key=int(self.address)
            )
            self.img_acquirer.start(run_as_thread=True)
            self.logger.debug(self.cam_proto)