actfw_core package

Subpackages

Submodules

actfw_core.application module

class actfw_core.application.AppSettings(settings: Dict[str, Any], schema: Dict[str, actfw_core.application.SettingSchema])

Bases: object

class actfw_core.application.Application(stop_by_signals: typing.Iterable[signal.Signals] = (<Signals.SIGINT: 2>, <Signals.SIGTERM: 15>))

Bases: object

get_app_settings(path: str = 'setting_schema.json') → actfw_core.application.AppSettings

Get application settings from schema. :param path: path to schema file :type path: str

Returns

application settings

Return type

AppSettings

get_settings(default_settings: Dict[str, Any]) → Dict[str, Any]

Get given Act settings.

Parameters

default_settings (dict) – default settings

Returns

updated settings

Return type

dict

Notes

Copy default_settings and overwrite it by given Act settings.

register_task(task: actfw_core.task.task.Task) → None

Register the application task.

Parameters

task (Task) – task

run() → None

Start application

running: bool
settings: Optional[Dict[str, Any]]

Actcast Application

stop() → None

Stop application

tasks: List[actfw_core.task.task.Task]
class actfw_core.application.SettingSchema(title: str, description: str, type_: type, default: Any = None, ui_type: Optional[str] = None)

Bases: object

static decoder(obj: Any) → Any
static infertype(typestring: str) → type

actfw_core.autofocus module

class actfw_core.autofocus.AfMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

AfModeAuto = 1
AfModeContinuous = 2
AfModeManual = 0
class actfw_core.autofocus.AfPause(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

AfPauseDeferred = 1
AfPauseImmediate = 0
AfPauseResume = 2
class actfw_core.autofocus.AfPauseState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

Paused = 2
Pausing = 1
Running = 0
class actfw_core.autofocus.AfRange(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

AfRangeFull = 'full'
AfRangeMacro = 'macro'
AfRangeMax = 'max'
AfRangeNormal = 'normal'
class actfw_core.autofocus.AfSpeed(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

AfSpeedFast = 'fast'
AfSpeedMax = 'max'
AfSpeedNormal = 'normal'
class actfw_core.autofocus.AfState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

Failed = 3
Focused = 2
Idle = 0
Scanning = 1
class actfw_core.autofocus.AfStatus

Bases: object

class actfw_core.autofocus.AutoFocuserBase(afrange: actfw_core.autofocus.AfRange = AfRange.AfRangeNormal, afspeed: actfw_core.autofocus.AfSpeed = AfSpeed.AfSpeedNormal, afmode: actfw_core.autofocus.AfMode = AfMode.AfModeContinuous)

Bases: object

Autofocus control algorithm https://github.com/raspberrypi/libcamera/blob/v0.3.2%2Brpt20240927/src/ipa/rpi/controller/rpi/af.cpp

get_focus_stats() → actfw_core.autofocus.AfStatus

Return latest focus status.

get_focus_value() → Optional[int]

Return latest focus value.

abstract parse_pdaf(meta_buffer: actfw_core.v4l2.video.VideoBuffer) → actfw_core.autofocus.PdafRegions
parse_pdaf_and_update_focus(meta_buffer: actfw_core.v4l2.video.VideoBuffer) → None

Parse PDAF data from embedded metadata from the unicam meta device. Calculate the optimal lens position based on the analyzed data, and execute the callback function.

process_contrast_metadata(stats: actfw_core.v4l2.types.bcm2835_isp_stats) → None

Extract contrast information from the metadata received from the ISP device and store it for later use.

set_focus_value(value: int) → None
set_focus_windows(bboxes: List[List[float]]) → None

Configures the focus windows. bboxes: [[x1, y1, x2, y2], …] defined in capture image coordinate system.

set_unicam_config(sensor_config: Dict, unicam_width: int, sensor_size: Tuple[int, int], post_crop_size: Tuple[int, int, int, int], camera_mode_scale: Tuple[float, float], camera_mode_crop_size: Tuple[int, int], callback_fn: Callable) → None

This method is called in UnicamIspCapture.__init__().

trigger_scan() → None
class actfw_core.autofocus.AutoFocuserIMX708(afrange: actfw_core.autofocus.AfRange = AfRange.AfRangeNormal, afspeed: actfw_core.autofocus.AfSpeed = AfSpeed.AfSpeedNormal, afmode: actfw_core.autofocus.AfMode = AfMode.AfModeContinuous)

Bases: actfw_core.autofocus.AutoFocuserBase

parse_pdaf(meta_buffer: actfw_core.v4l2.video.VideoBuffer) → actfw_core.autofocus.PdafRegions
class actfw_core.autofocus.CfgParams(params: Dict[str, Any])

Bases: object

read(params: Dict[str, Any]) → None
class actfw_core.autofocus.PdafData(conf: int, phase: int)

Bases: object

class actfw_core.autofocus.PdafRegions(height: int, width: int)

Bases: object

class actfw_core.autofocus.Pwl(points: List[float])

Bases: object

eval(x: float) → float
findSpan(x: float, span: int) → int

Pwls are generally small, so linear search may well be faster than binary, though could review this if large Pwls start turning up.

inverse_eval(y: int) → Optional[float]
class actfw_core.autofocus.RangeDependentParams

Bases: object

read(params: Dict[str, Any]) → None
class actfw_core.autofocus.Rectangle(xpos: float, ypos: float, w: float, h: float)

Bases: object

class actfw_core.autofocus.RegionWeights(rows: int, cols: int)

Bases: object

class actfw_core.autofocus.ScanRecord(focus: float, contrast: float, phase: float, conf: float)

Bases: object

class actfw_core.autofocus.ScanState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

Coarse = 3
Fine = 4
Idle = 0
Pdaf = 2
Settle = 5
Trigger = 1
class actfw_core.autofocus.SpeedDependentParams

Bases: object

read(params: Dict[str, Any]) → None
actfw_core.autofocus.clamp(x: float, low: float, high: float) → float

actfw_core.capture module

class actfw_core.capture.Frame(value: actfw_core.capture.T)

Bases: Generic[actfw_core.capture.T]

getvalue() → actfw_core.capture.T

Get frame data.

Returns

captured image data

Return type

bytes

value: actfw_core.capture.T

Captured Frame

class actfw_core.capture.V4LCameraCapture(device: typing.Union[str, actfw_core.system.DeviceInfo] = '/dev/video0', size: typing.Tuple[int, int] = (640, 480), framerate: int = 30, expected_format: actfw_core.v4l2.video.V4L2_PIX_FMT = V4L2_PIX_FMT.RGB24, fallback_formats: typing.Iterable[actfw_core.v4l2.video.V4L2_PIX_FMT] = (<V4L2_PIX_FMT.YUYV: 1448695129>, <V4L2_PIX_FMT.MJPEG: 1196444237>), format_selector: actfw_core.capture.FormatSelector = FormatSelector.DEFAULT)

Bases: actfw_core.task.producer.Producer[actfw_core.capture.Frame[bytes]]

Parameters

  • device (str | DeviceInfo) – v4l device path or info

  • size (int, int) – expected capture resolution

  • framerate (int) – expected capture framerate

  • expected_format (V4L2_PIX_FMT) – expected capture format

  • fallback_formats (list of V4L2_PIX_FMT) – fallback capture format

  • ( (format_selector) – class:`~actfw_core.capture.V4LCameraCapture.FormatSelector): how to select a format from listed formats supported by a camera. # noqa: B006 B950 DEFAULT selects the first format that meets the conditions. PROPER selects the smallest format that meets the conditions. MAXIMUM selects the largest resolution format as a camera can. MAXIMUM ignores framerate parameters (uses appropriate framerate for the selected resolution). If a camera lists [1280x720, 1920x1080, 640x480, 800x600] and an expected capture resolution is (512, 512), DEFAULT selects 1280x720, PROPER selects 800x600 and MAXIMUM selects 1920x1080.

Notes

If a camera doesn’t support the expected_format, try to capture one of the fallback_formats and convert it to expected_format.

class FormatSelector(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

Captured Frame Producer for Video4Linux

DEFAULT = 1
MAXIMUM = 3
PROPER = 2
capture_format: actfw_core.v4l2.video.V4L2_PIX_FMT
capture_height: int
capture_size() → Tuple[int, int]

Get configured capture resolution. A configured resolution may be more larger than expected one.

Returns

configured capture resolution (width, height)

Return type

(int, int)

capture_width: int
configure(configurator: Callable[[actfw_core.v4l2.video.Video], actfw_core.capture.CONFIGURATOR_RETURN]) → actfw_core.capture.CONFIGURATOR_RETURN

Run user defined video configurator.

Parameters

configurator – unary function (actfw_core.v4l2.video.Video -> a)

Returns

return type of configurator

Return type

object

run() → None

Run producer activity

video: actfw_core.v4l2.video.Video

actfw_core.command_server module

class actfw_core.command_server.CommandServer(sock_path: Optional[str] = None, custom_command_handler: Optional[Callable[[actfw_core.command_server.CustomCommandRequest], str]] = None)

Bases: actfw_core.task.isolated.Isolated

img: Optional[PIL.Image.Image]

Actcast Command Server

This server handles these commands

  • ‘Take Photo’

    • responses cached image as png data

img_lock: _thread.allocate_lock
run() → None

Run and start the activity

sock_path: Optional[str]
update_image(image: PIL.Image.Image) → None

Update the cached ‘Take Photo’ command image.

Parameters

image (Image) – image

class actfw_core.command_server.CustomCommandRequest

Bases: TypedDict

id: str
payload: str

actfw_core.libcamera_capture module

LibcameraCapture module.

Note

This module is only available in ActcastOS4 or later.

exception actfw_core.libcamera_capture.CameraConfigurationInvalidError(config: libcamera._libcamera.CameraConfiguration, msg: str = 'Invalid CameraConfiguration')

Bases: Exception

exception actfw_core.libcamera_capture.CameraConfigureError(errno: int, msg: str = 'Camera configure failed')

Bases: Exception

exception actfw_core.libcamera_capture.CameraStartError(errno: int, msg: str = 'CameraStart error')

Bases: Exception

exception actfw_core.libcamera_capture.CaptureTimeoutError(timeout: int, msg: str = 'Capture timeout')

Bases: Exception

exception actfw_core.libcamera_capture.FrameBufferAllocateError(errno: int, msg: str = 'FrameBuffer allocation error')

Bases: Exception

class actfw_core.libcamera_capture.LibcameraCapture(size: typing.Tuple[int, int], pixel_format: libcamera._libcamera.PixelFormat, camera_index: int = 0, orientation: libcamera._libcamera.Orientation = <Orientation.Rotate0: 1>, framerate: int = 30, depad: bool = True)

Bases: actfw_core.task.producer.Producer[actfw_core.capture.Frame[bytes]]

Initialization method for the LibcameraCapture class.

Parameters

  • size – The size of the captured image (width, height).

  • pixel_format – The pixel format. Only RGB888 or BGR888 are supported. cf. https://libcamera.org/api-html/classlibcamera_1_1PixelFormat.html

  • camera_index – The index of the camera to use. Defaults to 0. See find_csi_camera_index() to select the CSI camera when a USB camera is also connected.

  • orientation – The orientation of the camera. Defaults to Rotate0. cf. https://libcamera.org/api-html/namespacelibcamera.html#a80ea01625b93ecfe879249ac60c79384.

  • framerate – The framerate (fps). If not specified, the default setting of the libcamera is used.

  • depad – Whether to strip the per-line stride padding from captured frames. Defaults to True. The ISP may pad each line of the output buffer up to an alignment boundary (e.g. 64 bytes on Raspberry Pi 5 / PiSP), so the stride can be larger than width * 3. When True, the padding is removed and a tightly packed (width * 3 per line) buffer is produced, so consumers can read it as a plain width x height x 3 image regardless of the capture width. When False, the raw buffer is passed through unchanged; in that case use stride() to interpret it yourself (this avoids the per-line depadding repack and is useful when you can handle padded strides).

Note

As for pixel_format, if RGB888 is specified, BGR888 is actually obtained,

and if BGR888 is specified, RGB888 is actually obtained.

Raises
cameras() → List[libcamera._libcamera.Camera]
capture_size() → Tuple[int, int]
run() → None

Run and start the activity

set_exposure_settings(shutter_time: Union[float, actfw_core.unicam_isp_capture.Auto], analogue_gain: Union[float, actfw_core.unicam_isp_capture.Auto]) → None

Set shutter_time and analogue_gain.

Must be called before the capture task is started (i.e. before run()), since the controls are applied at camera.start().

Parameters

  • shutter_time (float or Auto) – [μsec] (e.g. 60000). If set to Auto, shutter_time is set with AE algorithm.

  • analogue_gain (float or Auto) – (>=1.0) (e.g. 2.5). If set to Auto, analogue_gain is set with AE algorithm.

stride() → int

Return the stride (bytes per line) of the captured buffer, including padding.

The ISP may align each line of the output buffer up to a boundary (e.g. 64 bytes on Raspberry Pi 5 / PiSP), so this can be larger than width * 3. When the capture is created with depad=False, use this together with capture_size() to interpret the raw frame buffer (e.g. Image.frombuffer(..., "raw", "RGB", stride, 1) for PIL, or numpy.lib.stride_tricks.as_strided for a zero-copy ndarray view).

exception actfw_core.libcamera_capture.QueueRequestError(errno: int, msg: str = 'QueueRequest error')

Bases: Exception

actfw_core.libcamera_capture.find_csi_camera_index() → Optional[int]

Return the index of the first CSI camera, or None if there is none. (Raspberry Pi specific.)

Intended for selecting a single CSI camera. A CSI camera is identified by its platform/ id prefix (set by the rpi/pisp / rpi/vc4 pipeline handler); the location property cannot be used since Raspberry Pi CSI cameras report Location = External just like USB ones. Pass the result to LibcameraCapture(..., camera_index=...) to avoid accidentally picking a USB camera that libcamera may also enumerate.

Note

The returned index is not a stable id; it is a position in the libcamera enumeration order and may differ between runs, so use it promptly rather than persisting it.

actfw_core.local_video_server module

class actfw_core.local_video_server.LocalVideoServer(quality: int = 75)

Bases: actfw_core.task.isolated.Isolated

Initialize a LocalVideoServer instance.

Parameters

quality (int, optional) –

The image quality setting. Acceptable values range from 0 to 100.

  • 0: Lowest quality

  • 95: Highest quality

  • 100: JPEG lossless (no compression)

The quality parameter corresponds to the quality parameter in PIL’s Image.save method. Defaults to 75.

image: actfw_core.local_video_server._ObservableValue[PIL.Image.Image]
run() → None

Run and start the activity

server: actfw_core.local_video_server._LocalVideoStreamServer

Local Video Server

This server provides a local video stream on port 5100.

stop() → None

Stop the activity

update_image(image: PIL.Image.Image) → None

Update the video image.

Parameters

image (Image) – image

actfw_core.service_client module

class actfw_core.service_client.ServiceClient(socket_path: Optional[pathlib.Path] = None)

Bases: object

rs256(payload: bytes) → str

Sign a message with an actcast device specific secret key.

Parameters

payload (bytes) – message

Returns

signature (base64url encoded)

Return type

str

Exceptions:

RuntimeError

stop_act() → NoReturn

Request actcast agent to stop the act.

Exceptions:

RuntimeError

actfw_core.system module

class actfw_core.system.DeviceInfo(type: str, nodes: List[actfw_core.system.DeviceNode])

Bases: object

static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceInfo
nodes: List[actfw_core.system.DeviceNode]
type: str
class actfw_core.system.DeviceNode(path: pathlib.Path, label: Optional[str] = None)

Bases: object

static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceNode
label: Optional[str] = None
path: pathlib.Path
class actfw_core.system.DeviceSupply(devices: List[actfw_core.system.DeviceInfo])

Bases: object

devices: List[actfw_core.system.DeviceInfo]
static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceSupply
exception actfw_core.system.EnvironmentVariableNotSet(name: str)

Bases: Exception

actfw_core.system.find_csi_camera_device() → Optional[str]

Path of CSI camera device, or None if no CSI camera is connected. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

Note

On Raspberry Pi 5, the returned path is an rp1-cfe (RP1 Camera Front End) node, which is a raw CSI-2 receiver configured through libcamera’s Media Controller pipeline. It is suitable for checking CSI camera presence and for the V4L-based capture path on older devices, but on Raspberry Pi 5 the capture itself should go through LibcameraCapture (which selects a camera by index, not by this V4L2 node). See actfw_core.libcamera_capture.find_csi_camera_index.

actfw_core.system.find_usb_camera_device() → Optional[str]

Path of USB camera device. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_act_settings_path() → str

Path of act settings. This is rather low level API. Use get_settings in Application to get act settings. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_act_id() → int

Act ID of the actcast application. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_agent_simulator() → Optional[str]

Fixed value “actsim” if on actsim environment. Not set otherwise. Since ACTCAST_PROTOCOL_VERSION 1.1.0.

actfw_core.system.get_actcast_command_sock() → str

Path of socket file to receive command from actcast agent. This is rather low level API. Use CommandServer to receive commands from actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_device_id() → str

Device ID of the device running the actcast application. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_device_type() → str

Device type. Since ACTCAST_PROTOCOL_VERSION 1.1.0.

actfw_core.system.get_actcast_firmware_type() → str

Firmware type of the host. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_actcast_group_id() → int

Group ID that the device is belonging to. Since ACTCAST_PROTOCOL_VERSION 1.2.0.

actfw_core.system.get_actcast_instance_id() → str

ID that identifies the launch of the actcast application, like PID. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_protocol_version() → str

Protocol version of agent-app communication. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_service_sock() → str

Path of socket file to send command to actcast agent. This is rather low level API. Use ServiceClient to send commands to actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_socks_server() → Optional[str]

URL of SOCKS5 proxy server. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_camera_device_info(default_image_source: Optional[str] = None) → actfw_core.system.DeviceInfo

DeviceInfo for camera. Set default_image_source only if you write camera device path in manifesto files. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_device_supply() → actfw_core.system.DeviceSupply

Device Supply from actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_device_supply_path() → str

Path of device supply file. This is rather low level API. Use get_device_supply to get DeviceSupply object. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.unicam_isp_capture module

class actfw_core.unicam_isp_capture.Auto(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

AUTO = 1
class actfw_core.unicam_isp_capture.CameraMode(size: Tuple[int, int], scale: Tuple[float, float], crop: Tuple[int, int])

Bases: object

crop: Tuple[int, int]
scale: Tuple[float, float]
size: Tuple[int, int]
class actfw_core.unicam_isp_capture.UnicamIspCapture(unicam: str = '/dev/video0', unicam_meta: str = '/dev/video1', unicam_subdev: str = '/dev/v4l-subdev0', unicam_subdev_meta: str = '/dev/v4l-subdev1', isp_in: str = '/dev/video13', isp_out_high: str = '/dev/video14', isp_out_metadata: str = '/dev/video16', size: Tuple[int, int] = (640, 480), unicam_size: Optional[Tuple[int, int]] = None, crop_size: Optional[Tuple[int, int, int, int]] = None, framerate: int = 30, expected_format: actfw_core.v4l2.video.V4L2_PIX_FMT = V4L2_PIX_FMT.RGB24, auto_whitebalance: bool = True, init_controls: List[str] = [], agc: bool = True, target_Y: float = 0.16, alsc: bool = True, default_color_temperature: int = 4500, contrast: bool = True, config: Optional[Dict[str, Any]] = None, vflip: bool = False, hflip: bool = False, shutter_time: Union[float, actfw_core.unicam_isp_capture.Auto] = Auto.AUTO, analogue_gain: Union[float, actfw_core.unicam_isp_capture.Auto] = Auto.AUTO, auto_focuser: Optional[actfw_core.autofocus.AutoFocuserBase] = None)

Bases: actfw_core.task.producer.Producer[actfw_core.capture.Frame[bytes]]

capture_size() → Tuple[int, int]
run() → None

Run and start the activity

set_exposure_settings(shutter_time: Union[float, actfw_core.unicam_isp_capture.Auto], analogue_gain: Union[float, actfw_core.unicam_isp_capture.Auto]) → None

Set shutter_time and analogue_gain.

Parameters

  • shutter_time (float or Auto) – [μsec] (e.g. 60000). If set to Auto, shutter_time is set with agc algorithm .

  • analogue_gain (float or Auto) – (>=0) (e.g. 2.5). If set to Auto, analogue_gain is set with agc algorithm .

set_exposure_time(ms: Optional[int] = None) → bool

Set exposure time.

This function is no longer supported. Use set_exposure_settings instead.

Returns

False

Module contents

actfw_core.heartbeat(*args: Any, **kwargs: Any) → None

Execute heartbeat action.

Notes

Default action is ‘touch /root/heartbeat’.

actfw_core.notify(notification: List[Dict[str, Any]], *args: Any, **kwargs: Any) → None

Make a notification to Actcast.

Parameters

notification (list of dict) – dicts must be encodable to JSON.

Example

>>> import actfw_core
>>> actfw_core.notify([{'msg': 'Hello!'}])
[{"msg": "Hello!"}]
actfw_core.set_heartbeat_function(f: Callable[[...], None]) → None

Set heartbeat action.

Parameters

f (function) – function which execute heartbeat action

Example

>>> import actfw_core
>>> def heartbeat(): print("working!")
...
>>> actfw_core.set_heartbeat_function(heartbeat)
>>> actfw_core.heartbeat()
working!