Client

pycertifspec.Client #

[view_source]

Client Objects # 

class Client()

[view_source]

Connection to SPEC

You should only need one instance of this class and use it to create Motor, Variable, etc. instances

__init__ # 

 | __init__(host: str = "localhost", port: int = None, port_range: Tuple[int, int] = (6510, 6530), ports: List[int] = [], timeout: float = 0.5, log_messages: bool = False)

[view_source]

Attempt to create a connection to SPEC

Attributes:

  • host string - The address of the SPEC server
  • port int - If exact port is known, the port to connect to
  • port_range tuple[int] - Range of ports to scan (end is inclusive)
  • ports list[int] - List of ports to scan
  • timeout float - Time to wait for answer before trying the next port
  • log_messages boolean - Print all incoming SpecMessages

subscribe # 

 | subscribe(prop: str, callback: Callable[[SpecMessage], None], nowait: bool = False, timeout: float = 0.2) -> bool

[view_source]

Subscribe to changes in a property.

Arguments:

  • prop string - The name of the property
  • callback function - The function to be called when the event is received. Will also be called immediately after subscribing
  • nowait boolean - By default the function waits for the first event after registering to see if an error occurred. To skip that set True
  • timeout float - The timeout to wait for a response after subscribing. Function returns False when it runs out

Returns:

True if successful, False when timeout reached

unsubscribe # 

 | unsubscribe(prop: str, callback: Callable[[SpecMessage], None]) -> bool

[view_source]

Unsubscribe from changes in the property.

Arguments:

  • prop string - To property to unsubscribe from
  • callback function - The callback function

Returns:

  • (boolean) - True if the callback was removed, False if it didn’t exist anyways

run # 

 | run(console_command: str, blocking: bool = True, callback: Callable[[SpecMessage, str], None] = None) -> Tuple[SpecMessage, str]

[view_source]

Execute a command like from the interactive SPEC console

Arguments:

  • console_command string - The command to execute
  • blocking boolean - When True, the function will block until it receives a response from SPEC and return the response
  • callback function - When blocking=False, the response will instead be send to the callback function. Expected to accept 2 positional arguments: data, console_output

Returns:

Tuple[SpecMessage, str]: If blocking, the response message from the server and what would be printed to console

set # 

 | set(prop: str, value: Any, wait_for_error: float = 0.2)

[view_source]

Set a property.

Attributes:

  • prop_name string - The name of the property
  • value - The value (will be converted to string before sending)
  • wait_for_error float - SPEC only sends a message back if the property doesn’t exist. Set the number of seconds to wait for an error message (if there is one)

get # 

 | get(prop: str, force_fetch: bool = False) -> SpecMessage

[view_source]

Get a property.

Attributes:

  • prop_name string - The name of the property
  • force_fetch boolean - If the property is watched, force fetch the value from SPEC

Returns:

  • (SpecMessage) - None if property doesn’t exist. Else SpecMessage

watch # 

 | watch(prop: str) -> bool

[view_source]

Listen for changes in prop to speed up .get() method

Arguments:

  • prop string - Name of the prop to watch

Returns:

  • (boolean) - True if successful

unwatch # 

 | unwatch(prop: str)

[view_source]

Stop listening for changes in prop

Arguments:

  • prop string - Name of the prop to stop watching

motor # 

 | motor(mne: str) -> Motor

[view_source]

Get the motor as an object

Arguments:

  • mne string - The mnemonic name of the motor

Returns:

  • (Motor) - The motor

var # 

 | var(name: str, dtype: Type = str) -> Union[Var, ArrayVar]

[view_source]

Get the variable as an object

Arguments:

  • name string - The name of the variable
  • dtype Type - The type of the variable

Returns:

(Var or ArrayVar): The variable

abort # 

 | abort()

[view_source]

Abort all running commands

motors # 

 | @property
 | motors() -> List[str]

[view_source]

List of all available motor mnemonic names

Returns:

  • (List[str]) - List of all motor names

motor_names # 

 | @property
 | motor_names() -> 'OrderedDict[str, str]'

[view_source]

Dict of all available motor mnemonic and pretty names

Returns:

(OrderedDict[str, str]): Dict with mnemonic names as keys and pretty names as values

count # 

 | count(time: float, callback: Callable = None, refresh_names: bool = False) -> Dict[str, float]

[view_source]

Counts scalers for the time specified. This function is blocking. The callback function will receive occasional updates during counting and when counting is finished.

Arguments:

  • time float - The time to count in seconds
  • callback function - Callback function for updates during counting
  • refresh_names boolean - If True, counter names will be refreshed before starting to count. Only necessary if a counter has been added or removed since the script started.

Returns:

(Dict[str, float]): Counter values

stop_counting # 

 | stop_counting()

[view_source]

Stop counting immediately. Will also cause .count() call to return if started in different thread.