actfw_core package

Subpackages

Submodules

actfw_core.application module

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

Bases: object

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]

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.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.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_subdev: str = '/dev/v4l-subdev0', 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)

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

adjust_setting_from_isp() None
agc(stats: actfw_core.v4l2.types.bcm2835_isp_stats) None
alsc() None
apply_ls_tables() None
auto_size_selection_for_imx219() None
auto_size_selection_for_ov5647() None
awb(stats: actfw_core.v4l2.types.bcm2835_isp_stats) None
calc_crop_size(isp_out_width: int, isp_out_height: int, unicam_width: int, unicam_height: int) Tuple[int, int, int, int]
calc_lux(isp_stats: actfw_core.v4l2.types.bcm2835_isp_stats) None
calculate_y(stats: actfw_core.v4l2.types.bcm2835_isp_stats, additional_gain: float) float
capture_size() Tuple[int, int]
compose_gamma_curve(one: List[Tuple[float, float]], other: List[Tuple[float, float]], eps: float = 1e-06) List[Tuple[float, float]]
compute_stretch_curve(histogram: List[int]) List[Tuple[float, float]]
contrast_control(isp_stats: actfw_core.v4l2.types.bcm2835_isp_stats) None
cumulative_quantile(cumulative: List[int], q: float, first: int = - 1, last: int = - 1) float
eval_gamma_curve(gamma_curve: List[Tuple[float, float]], x: float) float
fill_in_contrast_status(gm: actfw_core.v4l2.types.bcm2835_isp_gamma, gamma_curve: List[Tuple[float, float]]) None
find_span(gamma_curve: List[Tuple[float, float]], x: float) int
get_cal_table(ct: float, calibrations: List[Dict[str, Any]]) List[float]
get_sensor_name(subdev: str) str
histogram_cumulative(histogram: List[int]) List[int]
isp2unicam() None
populate_ls_table(src: List[float], dst_w: int, dst_h: int, ls_table: mmap.mmap, offset: int) None
produce_image_from_isp() None
request_buffer() None
resample_cal_table(src: List[float], camera_mode: actfw_core.unicam_isp_capture.CameraMode) List[float]
run() None

Run and start the activity

set_unicam_exposure(analogue_gain: float, shutter_time: float) None
set_unicam_fps() None
setup_pipeline() None
unicam2isp() None
v4l2_control_value_for_analogue_gain(analogue_gain: float) float

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!