MockServer¶

class lsst.ts.cbp.MockServer(log=None)¶

Bases: OneClientServer

Mocks the CBP server.

Parameters:

loglogging.Logger, optional

Attributes:

hoststr
portint
timeoutint
long_timeoutint
azimuthfloat
altitudefloat
focusint
maskstr
panic_statusbool
encodersEncoders
parkbool
auto_parkbool
masks_rotationdict of str:float
commandstuple of re.Pattern:functools.partial
loglogging.Logger

Attributes Summary

connected

Return True if self.reader and self.writer are connected.

Methods Summary

`basic_close_client`()	Close the connected client socket, if any.
`call_connect_callback`()	Call self.__connect_callback.
`close`()	Close socket server and client socket, and set done_task done.
`close_client`()	Close the connected client socket, if any.
`cmd_loop`()	Run the command loop.
`connect_callback`(server)
`do_aastat`()	Return the azimuth encoder status.
`do_abstat`()	Return the altitude encoder status.
`do_acstat`()	Return the focus encoder status.
`do_adstat`()	Return the mask selection encoder status.
`do_aestat`()	Return the mask rotation encoder status.
`do_altitude`()	Return the altitude position.
`do_autopark`()	Return the autopark value.
`do_azimuth`()	Return azimuth position.
`do_focus`()	Return the focus value.
`do_mask`()	Return the mask value.
`do_new_altitude`(altitude)	Set the new altitude position.
`do_new_azimuth`(azimuth)	Set the new azimuth position.
`do_new_focus`(focus)	Set the new focus value.
`do_new_mask`(mask)	Set the new mask value.
`do_new_rotation`(rotation)	Set the new mask rotation value.
`do_panic`()	Return the panic status value.
`do_park`([park])	Park or unpark the CBP.
`do_rotation`()	Return the mask rotation value.
`read`(n)	Read up to n bytes.
`read_into`(struct)	Read binary data from a stream reader into a `ctypes.Structure`.
`readexactly`(n)	Read exactly n bytes.
`readline`()	Read a sequence of bytes ending with `\n`.
`readuntil`([separator])	Read one line, where “line” is a sequence of bytes ending with
`set_constrained_position`(value, actuator)	Set actuator to position that is silently constrained to bounds.
`start`(**kwargs)	Start the TCP/IP server.
`write`(data)	Write data and call `drain`.
`write_from`(*structs)	Write binary data from one or more `ctypes.Structure`s.
`writelines`(lines)	Write an iterable of bytes and call `drain`.

Attributes Documentation

connected¶

Return True if self.reader and self.writer are connected.

Note: if the other end drops the connection and if you are not trying to read data (e.g. in a background loop), then it takes the operating system awhile to realize the connection is lost. So this can return true for some unknown time after the connection has been dropped.

Methods Documentation

async basic_close_client() → None¶

Close the connected client socket, if any.

Also:

Reset self.connected_task to a new Future.
Call connect_callback, if a client was connected.

Unlike close_client, this does not touch self.should_be_connected.

Always safe to call.

async call_connect_callback() → None¶

Call self.__connect_callback.

This is always safe to call. It only calls the callback function if that function is not None and if the connection state has changed since the last time this method was called.

async close() → None¶

Close socket server and client socket, and set done_task done.

Call connect_callback if a client was connected.

Always safe to call.

async close_client() → None¶

Close the connected client socket, if any.

Also:

Set self.should_be_connected false.
Reset self.connected_task to a new Future.
Call connect_callback, if a client was connected.

Always safe to call.

async cmd_loop()¶

Run the command loop.

Parameters:

readerasyncio.StreamReader
writerasyncio.StreamWriter

connect_callback(server)¶

async do_aastat()¶

Return the azimuth encoder status.

Returns:

str

async do_abstat()¶

Return the altitude encoder status.

Returns:

str

async do_acstat()¶

Return the focus encoder status.

Returns:

str

async do_adstat()¶

Return the mask selection encoder status.

Returns:

str

async do_aestat()¶

Return the mask rotation encoder status.

Returns:

str

async do_altitude()¶

Return the altitude position.

Returns:

str

async do_autopark()¶

Return the autopark value.

Returns:

str

async do_azimuth()¶

Return azimuth position.

Returns:

str

async do_focus()¶

Return the focus value.

Returns:

str

async do_mask()¶

Return the mask value.

Returns:

str

async do_new_altitude(altitude)¶

Set the new altitude position.

Parameters:

altitudefloat

Returns:

str

async do_new_azimuth(azimuth)¶

Set the new azimuth position.

Parameters:

azimuthfloat

Returns:

str

async do_new_focus(focus)¶

Set the new focus value.

Parameters:

focus

Returns:

str

async do_new_mask(mask)¶

Set the new mask value.

Parameters:

maskstr

Returns:

str

async do_new_rotation(rotation)¶

Set the new mask rotation value.

Parameters:

rotationfloat

Returns:

str

async do_panic()¶

Return the panic status value.

Returns:

str

async do_park(park='?')¶

Park or unpark the CBP.

Parameters:

parkint, optional

Returns:

str

async do_rotation()¶

Return the mask rotation value.

Returns:

str

async read(n: int) → bytes¶

Read up to n bytes.

Parameters:

nint: The number of bytes to read. If -1 then block until the other end closes its writer, then return all data seen.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async read_into(struct: Structure) → None¶

Read binary data from a stream reader into a ctypes.Structure.

Parameters:

structctypes.Structure: Structure to set.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async readexactly(n: int) → bytes¶

Read exactly n bytes.

Parameters:

nint: The number of bytes to read.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async readline() → bytes¶

Read a sequence of bytes ending with \n.

If EOF is received and \n was not found, the method returns partially read data.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async readuntil(separator: bytes = b'\n') → bytes¶

Read one line, where “line” is a sequence of bytes ending with .

Read data from the stream until separator is found.

On success, the data and separator will be removed from the internal buffer (consumed). Returned data will include the separator at the end.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.

set_constrained_position(value, actuator)¶

Set actuator to position that is silently constrained to bounds.

Parameters:

valuefloat: Desired value
actuatorlsst.ts.simactuators.PointToPointActuator: The actuator to set.

async start(**kwargs: Any) → None¶

Start the TCP/IP server.

This is called automatically by the constructor, and is not intended to be called by the user. It is a public method so that subclasses can override it.

Parameters:

**kwargsdict [str, typing.Any]: Additional keyword arguments for asyncio.start_server, beyond host and port.

Raises:

RuntimeError: If start has already been called and has successfully constructed a server.

async write(data: bytes) → None¶

Write data and call drain.

Parameters:

databytes: The data to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async write_from(*structs: Structure) → None¶

Write binary data from one or more ctypes.Structures.

Parameters:

structctypes.Structure: Structure to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async writelines(lines: Iterable) → None¶

Write an iterable of bytes and call drain.

Parameters:

linescollections.abc.Iterable [bytes]: The data to write, as an iterable collection of bytes.

Raises:

ConnectionError: If self.connected false before writing begins.

Navigation

MockServer¶