Coverage for core / src / sensorkit / cli / controller.py: 0%

1# SPDX-License-Identifier: Apache-2.0 

2import datetime as dt 

3import json 

4 

5import asyncclick as click 

6 

7from sensorkit.cli.utils import entity_option, with_kit 

8 

9 

10@click.group("controller") 

11async def controller_group(): 

12 """Controller management commands.""" 

13 

14@controller_group.command("abort") 

15@entity_option(required=True) 

16@with_kit 

17async def abort(kit, entity: str): 

18 """Abort the currently running task on a controller.""" 

19 await kit.controller(entity).abort_task() 

20 

21 

22@controller_group.command("init") 

23@entity_option(required=True) 

24@click.option("-f", "--force", is_flag=True, default=False) 

25@with_kit 

26async def startup(kit, entity: str, force: bool): 

27 """Initialize a controller.""" 

28 from sensorkit.core.task import InitTask 

29 

30 await kit.controller(entity).execute_task( 

31 InitTask(), 

32 expiry_time=dt.datetime.now(dt.UTC) + dt.timedelta(minutes=10), 

33 interrupt=force, 

34 ) 

35 

36 

37@controller_group.command("shutdown") 

38@entity_option(required=True) 

39@click.option("-f", "--force", is_flag=True, default=False) 

40@with_kit 

41async def shutdown(kit, entity: str, force: bool): 

42 """Shutdown a controller.""" 

43 from sensorkit.core.task import ShutdownTask 

44 

45 await kit.controller(entity).execute_task( 

46 ShutdownTask(), 

47 expiry_time=dt.datetime.now(dt.UTC) + dt.timedelta(minutes=10), 

48 interrupt=force, 

49 ) 

50 

51@controller_group.command("collect") 

52@entity_option(required=True) 

53@click.option("-t", "--target", required=True) 

54@click.option("-b", "--binning", type=int, default=1) 

55@click.option("-c", "--frame-count", type=int, default=1) 

56@click.option("-i", "--integration-time-seconds", type=float, default=1.0) 

57@with_kit 

58async def collect( 

59 kit, 

60 entity: str, 

61 target: str, 

62 binning: int, 

63 frame_count: int, 

64 integration_time_seconds: float, 

65): 

66 """Execute a StandardCollectTask on a controller with the given target and camera parameters.""" 

67 from sensorkit.std.collect import CameraParameterSet, StandardCollectTask 

68 

69 ctl = kit.controller(entity) 

70 

71 data = json.loads(target) 

72 collect_task = StandardCollectTask( 

73 target=data, 

74 camera_params=CameraParameterSet( 

75 integration_time_seconds=integration_time_seconds, 

76 frame_count=frame_count, 

77 binning_x=binning, 

78 binning_y=binning, 

79 ) 

80 ) 

81 

82 await ctl.execute_task( 

83 collect_task, 

84 expiry_time=dt.datetime.now(dt.UTC) + dt.timedelta(minutes=10), 

85 )