ugokukun.ugoku_kun

Control multiple cannon cameras and turntable.

class ugokukun.ugoku_kun.UgokuKun(task_csv_path: str, device_json_path: str, log_path: str = 'log.txt', wait_time: float = 0.5, max_attempts: int = 20, req_timeout: Tuple[float, float] = (3.0, 7.5))

Bases: object

Control multiple cannon cameras and turntable.

wait_time

Wait time between HTTP request attempts.

Type:

float

max_attempts

Maximum number of HTTP requests attempts.

Type:

int

req_timeout

Timeout of the HTTP request.

Type:

Tuple[float, float]

device_json_path

Path of the json file containing the device id and addresses.

Type:

str

devices

Dictionary containing the devices.

Type:

Dict[str, Dict[str, str]]

task_csv_path

Path of the task csv file.

Type:

str

csv_df

The csv file as a pandas DataFrame.

Type:

pd.DataFrame

log_path

Path of the log txt file.

Type:

str

logger

Logger object.

Type:

logging.Logger

keigan_motors

Dictionary containing the keigan motors.

Type:

Dict[str, str]

keigan_instances

Dictionary containing the KeiganWrapper instances.

Type:

Dict[str, KeiganWrapper]

cannon_cameras

Dictionary containing the cannon cameras.

Type:

Dict[str, str]

cannon_instances

Dictionary containing the CannonWrapper instances.

Type:

Dict[str, CannonWrapper]

execute_all(row: int)

Actions for target==”all”.

Parameters:

row (int) – Index of the row of the csv to be executed. Only action==”sleep” is implemeted.

Raises:

ValueError – Invalid action.

execute_cannon(camera: str, row_index: int) → Any

Execute a single row in the csv file where target==camera_id.

Parameters:

• camera (str) – camera_id in the device json file.

• row_index (int) – Index of the row of the csv to be executed.

Returns:

The return of the executed action. Most likely requests.Response object.

Return type:

Any

Raises:

ValueError – Invalid action.

Notes

if self.csv_df.at[row_index, “action”] is:

• “get”: self.cannon_instances[camera].http_get()

• “post”: self.cannon_instances[camera].http_post()

• “put”: self.cannon_instances[camera].http_put()

• “delete”: self.cannon_instances[camera].http_delete()

• “shutter”: self.cannon_instances[camera].shutter()

• “aperture”: self.cannon_instances[camera].set_aperture()

• “exposure”: self.cannon_instances[camera].set_exposure()

• “iso”: self.cannon_instances[camera].set_iso()

• “color_temperature”: self.cannon_instances[camera].set_color_temp()

• “white_balance”: self.cannon_instances[camera].set_white_balance()

• “shutter_speed”: self.cannon_instances[camera].set_shutter_speed()

• “sync_time”: self.cannon_instances[camera].sync_time()

• “dump”: self.cannon_instances[camera].dump_attributes()

execute_keigan(motor_id: str, row_index: int)

Execute a single row in the csv file where target==keigan_id.

Parameters:

• motor_id (str) – motor_id in the device json file.

• row_index (int) – Index of the row of the csv to be executed.

Returns:

The return of the executed action. Most likely None.

Return type:

Any

Raises:

ValueError – Invalid action.

execute_row(row_index: int) → None

Execure a single row in the csv file.

Parameters:

row_index (int) – Index of the row to be executed.

Raises:

ValueError – Invalid target.

load_task_csv(task_csv_path: str) → DataFrame

Load a new csv file and overwrite the current self.csv_df.

Parameters:

task_csv_path (str) – Path of the csv file.

Returns:

The csv file as a pandas DataFrame.

Return type:

pd.DataFrame

Raises:

ValueError – Missing header or duplicate task_id.

run() → None

Run all commands in the csv file.

test_run() → None

Run all commannds in the csv file except WAIT commands.

This method is useful to check if the commands are valid. Run this method before running the run method and fix if any errors are found.