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: Optional[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)

Bases: enum.Enum

An enumeration.

AfModeAuto = 1
AfModeContinuous = 2
AfModeManual = 0
class actfw_core.autofocus.AfPause(value)

Bases: enum.Enum

An enumeration.

AfPauseDeferred = 1
AfPauseImmediate = 0
AfPauseResume = 2
class actfw_core.autofocus.AfPauseState(value)

Bases: enum.Enum

An enumeration.

Paused = 2
Pausing = 1
Running = 0
class actfw_core.autofocus.AfRange(value)

Bases: enum.Enum

An enumeration.

AfRangeFull = 'full'
AfRangeMacro = 'macro'
AfRangeMax = 'max'
AfRangeNormal = 'normal'
class actfw_core.autofocus.AfSpeed(value)

Bases: enum.Enum

An enumeration.

AfSpeedFast = 'fast'
AfSpeedMax = 'max'
AfSpeedNormal = 'normal'
class actfw_core.autofocus.AfState(value)

Bases: enum.Enum

An enumeration.

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)

Bases: enum.Enum

An enumeration.

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)

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)

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

actfw_core.libcamera_capture module

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

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. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

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)

Bases: enum.Enum

An enumeration.

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!