""" Purpose: the Server provides proxy clients with methods for manipulating and querying the Plaxis environment and its global objects. The methods construct strings and call the Plaxis local server. The subsequent output is processed to create data for proxy client objects. This could take the form of a list of GUIDs or it could take the form of a string if requesting information about the state of the environment. If the request is not sucessful then a scripting exception is raised with the message that was returned from the interpreter. Subversion data: $Id: server.py 13274 2013-08-15 14:50:39Z ac $ $URL: http://tools.plaxis.com/svn/sharelib/trunk/PlxObjectLayer/Server/plxscripting/server.py $ Copyright (c) Plaxis bv. All rights reserved. Unless explicitly acquired and licensed from Licensor under another license, the contents of this file are subject to the Plaxis Public License ("PPL") Version 1.0, or subsequent versions as allowed by the PPL, and You may not copy or use this file in either source code or executable form, except in compliance with the terms and conditions of the PPL. All software distributed under the PPL is provided strictly on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESS OR IMPLIED, AND LICENSOR HEREBY DISCLAIMS ALL SUCH WARRANTIES, INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, QUIET ENJOYMENT, OR NON-INFRINGEMENT. See the PPL for specific """ from .plx_scripting_exceptions import PlxScriptingError from .const import (PLX_CMD_NEW, PLX_CMD_CLOSE, PLX_CMD_OPEN, PLX_CMD_RECOVER, JSON_COMMANDS, JSON_FEEDBACK, JSON_SUCCESS, JSON_EXTRAINFO, JSON_GUID, JSON_TYPE, JSON_RETURNED_OBJECTS, JSON_PROPERTIES, JSON_QUERIES, JSON_NAMEDOBJECTS, JSON_RETURNED_OBJECT, JSON_OWNERGUID, JSON_ISLISTABLE, METHOD, GUID, COUNT, JSON_LISTQUERIES, JSON_METHODNAME, JSON_OUTPUTDATA, SUBLIST, INDEX, STARTINDEX, STOPINDEX) from uuid import UUID try: basestring # will give NameError on Py3.x, but exists on Py2.x (ancestor of str and unicode) def isstr(s): return isinstance(s, basestring) except NameError: def isstr(s): # Py3.x only has one type of string return isinstance(s, str) plx_string_wrappers = ['"', "'", '"""', "'''"] class InputProcessor(object): """ Helper class which processes scripting input in order to present it correctly to the Plaxis server. """ def param_to_string(self, param): try: return param.get_cmd_line_repr() except AttributeError: # param has no appropriate method -> maybe it's a primitive if isstr(param): # strings need to wrapped with quotes (depending on whether they contain quotes themselves) for wrapper in plx_string_wrappers: if not wrapper in param: return wrapper + param + wrapper raise PlxScriptingError( "Cannot convert string parameter to valid Plaxis string representation, try removing some quotes: " + param) elif isinstance(param, (tuple, list)): # tuples/lists wrapped with parens return '(' + self.params_to_string(param) + ')' else: return str(param) def params_to_string(self, params): """ Takes a sequence and concatenates its contents into a single space separated string. E.g. params: (1, 2, 3) returns "1 2 3" params: () returns "" """ # TODO handle more complex cases such as the following: # * methods as params (as found when using the 'map' command) return " ".join([self.param_to_string(p) for p in params]) def create_method_call_cmd(self, target_object, method_name, params): """ Arranges the command line name of a proxy object, method name and parameters into a string with a format matching the Plaxis command line. E.g. target_obj_name: 'Point_1' method_name: "move" params: (3, 4, 5) returns "move Point_1 3 4 5" target_obj_name: "" (in the case of the global object) method_name: "undo" params: () returns "undo" """ param_string = self.params_to_string(params) parts = [method_name] if target_object is not None: target_cmd_line_repr = target_object.get_cmd_line_repr() if target_cmd_line_repr: parts.append(target_cmd_line_repr) if param_string: parts.append(param_string) return " ".join(parts) class ResultHandler(object): """ Helper class which parses the output of the Plaxis server and returns objects, primitives, booleans or strings if successful. Otherwise an exception is raised. """ def __init__(self, server, proxy_factory): self.proxy_factory = proxy_factory self.server = server def _handle_successful_command(self, commands_response): """ Supplied with a response from a successful command, returns one of the following if they are present: a list of proxy objects, extra information from the command line or True as a fallback. """ obj_list = self._create_proxies_from_returned_objects( commands_response[JSON_RETURNED_OBJECTS]) if obj_list: return obj_list else: json_extra_info = commands_response[JSON_EXTRAINFO] if json_extra_info: return json_extra_info return True def handle_namedobjects_response(self, namedobjects_response): """ Handles the JSON response to a call to the namedobjects resource. May return some newly created objects or a scripting error if the named object is not present. """ is_namedobject_successful = namedobjects_response[JSON_SUCCESS] if is_namedobject_successful: return self._create_proxies_from_returned_objects( [namedobjects_response[JSON_RETURNED_OBJECT]]) else: raise PlxScriptingError("Unsuccessful command:\n" + namedobjects_response[JSON_EXTRAINFO]) def handle_commands_response(self, commands_response): """ Handles the (JSON) response to a Plaxis method call. May return some newly created objects or some text indicating some change of state. If the method call is not successful, then an exception is raised. """ is_command_successful = commands_response[JSON_SUCCESS] if is_command_successful: return self._handle_successful_command(commands_response) else: raise PlxScriptingError("Unsuccessful command:\n" + commands_response[JSON_EXTRAINFO]) def handle_members_response(self, members_response, proxy_obj): """ Constructs and returns a dictionary containing the attribute names of an object mapped to the relevant proxy entity (either a proxy method or a proxy property). The supplied membernames response is the JSON object from the server that represents the attributes of the object. """ proxy_attributes = {} if JSON_COMMANDS in members_response: commands_list = members_response[JSON_COMMANDS] for method_name in commands_list: proxy_method = self.proxy_factory.create_plx_proxy_object_method( self.server, proxy_obj, method_name) proxy_attributes[method_name] = proxy_method if JSON_PROPERTIES in members_response: properties_dict = members_response[JSON_PROPERTIES] for property_name in properties_dict: ip = self._create_proxy_object( properties_dict[property_name], property_name, proxy_obj) proxy_attributes[property_name] = ip return proxy_attributes def handle_list_response(self, list_response): """ Handles the response to a call to the list resource. Depending on the call and the state of the project, the response may be a primitive, a proxy object, a list of proxy objects, or an error. """ is_listquery_successful = list_response[JSON_SUCCESS] if is_listquery_successful: method_name = list_response[JSON_METHODNAME] output_data = list_response[JSON_OUTPUTDATA] if method_name == COUNT: return output_data elif method_name == SUBLIST: return self._create_proxies_from_returned_objects(output_data) elif method_name == INDEX: return self._create_proxies_from_returned_objects( [output_data]) raise PlxScriptingError("Unsuccessful command:\n" + list_response[JSON_EXTRAINFO]) def handle_propertyvalues_response(self, propertyvalues_response, attr_name): """ Handle the request for a property. Returns the property. If there is no such attribute then the method returns None. """ # Make a call to Plaxis to get the object's properties if JSON_PROPERTIES in propertyvalues_response: property_names = propertyvalues_response[JSON_PROPERTIES] if attr_name in property_names: attribute = property_names[attr_name] # TODO: work out how to "proxify" the primitives if isinstance(attribute, dict): return self._create_proxies_from_returned_objects( [attribute]) return attribute def _create_proxy_object(self, returned_object, prop_name=None, owner=None): """ Accesses the data for a returned object and creates a proxy from that data. """ guid = returned_object[JSON_GUID] plx_obj_type = str(returned_object[JSON_TYPE]) # cast to str needed for Py2.7, where otherwise a potential unicode-type result object could lead to problems is_listable = returned_object[JSON_ISLISTABLE] # If we approach an object as listable, but its listification actually # returns intrinsic property objects that have NOT YET been cached as # proxies, we still need to identify them as intrprops and create the # appropriate proxy objects for them. if owner is None: # cannot simply write "if owner", because proxies may implement __bool__ and return False even if the owner does exist if JSON_OWNERGUID in returned_object: owner = self.proxy_factory.get_proxy_object_if_exists(returned_object[JSON_OWNERGUID]) # It shouldn't be possible to get a property back *before* we have # instantiated a proxy object for that property's owner. if owner is None: # also here cannot simply write "if owner" for similar reasons as above raise PlxScriptingError('Missing owner object for property object!') # The fact that the owner exists, doesn't necessarily mean all # its intrinsic properties have been retrieved too; and we need # them retrieved, because in the problematic situation we're # dealing with here, prop_name is *also* unknown and we can # therefore not instantiate the property ourselves! if self.proxy_factory.get_proxy_object_if_exists(guid) is None: # also here cannot simply write "if self.proxy_factory.get_proxy_object_if_exists" for similar reasons as above self.server.get_object_attributes(owner) return self.proxy_factory.create_plx_proxy_object( self.server, guid, plx_obj_type, is_listable, prop_name, owner) def _create_proxies_from_returned_objects(self, returned_objects): """ Given a returned objects list from the API, creates relevant proxy objects for each returned object representation. If the list contains just one object representation, a single proxy is returned. If the list contains more than one object representation, a list of proxies is returned. """ new_objs = [] for returned_object in returned_objects: obj = self._create_proxy_object(returned_object) new_objs.append(obj) if len(new_objs) == 1: return new_objs[0] return new_objs class Server(object): """ Provides proxy clients with the means to request and receive information from a connection to Plaxis. """ def __init__(self, connection, proxy_factory, input_processor): self.connection = connection self.input_proc = input_processor # TODO unsure about the tight coupling here self.plx_global = proxy_factory.create_plx_proxy_global(self) self.result_handler = ResultHandler(self, proxy_factory) self.__proxy_factory = proxy_factory def new(self): """ Create a new project """ res = self.connection.request_environment(PLX_CMD_NEW) if res: self.__proxy_factory.clear_proxy_object_cache() return res def recover(self): """ Recover a project """ res = self.connection.request_environment(PLX_CMD_RECOVER) if res: self.__proxy_factory.clear_proxy_object_cache() return res def open(self, filename): """ Open a project with the supplied name """ res = self.connection.request_environment(PLX_CMD_OPEN, filename) if res: self.__proxy_factory.clear_proxy_object_cache() return res def close(self): """ Close the current project """ res = self.connection.request_environment(PLX_CMD_CLOSE) if res: self.__proxy_factory.clear_proxy_object_cache() return res def call_listable_method(self, proxy_listable, method_name, startindex=None, stopindex=None): """ Constructs a listable query and returns the handled response. """ optional_parameters = {} # Attach any supplied index arguments to the query if startindex is not None: optional_parameters[STARTINDEX] = startindex if stopindex is not None: optional_parameters[STOPINDEX] = stopindex listable_query = {GUID: proxy_listable._guid, METHOD: method_name} listable_query.update(optional_parameters) response = self.connection.request_list(listable_query) return self.result_handler.handle_list_response( response[JSON_LISTQUERIES][0]) def get_named_object(self, object_name): """ Return a representation of the named object. """ response = self.connection.request_namedobjects(object_name) return self.result_handler.handle_namedobjects_response( response[JSON_NAMEDOBJECTS][object_name]) def get_object_property(self, proxy_object, prop_name, phase_object=None): """ Gets the specified property value for the specified proxy object. """ if phase_object: response = self.connection.request_propertyvalues( proxy_object._guid, prop_name, phase_object._guid) else: response = self.connection.request_propertyvalues( proxy_object._guid, prop_name) return self.result_handler.handle_propertyvalues_response( response[JSON_QUERIES][proxy_object._guid], prop_name) def set_object_property(self, proxy_property, prop_value): """ Sets the specified property value for the specified proxy object. """ return self.call_plx_object_method(proxy_property, 'set', prop_value) def get_object_attributes(self, proxy_obj): """ Create a dictionary of object attributes mapped to their proxy equivalents (proxy methods or proxy properties) """ response = self.connection.request_members(proxy_obj._guid) return self.result_handler.handle_members_response( response[JSON_QUERIES][proxy_obj._guid], proxy_obj) def call_plx_object_method(self, proxy_obj, method_name, params): """ Calls a Plaxis method using the supplied proxy object, method name and parameters. Returns new objects, success infomation, or a boolean if the command succeeds. Otherwise a scripting error is raised with any error information. E.g. proxy_obj: a Point object method_name: "move" params: (1, 2, 3) returns "OK" (from Plaxis command line) E.g. proxy_obj: the global proxy object method_name: "line" params: (1, 1, 1, 0, 0, 0) returns a list of PlxProxyObjects """ method_call_cmd = self.input_proc.create_method_call_cmd( proxy_obj, method_name, params) return self.call_and_handle_command(method_call_cmd) def call_and_handle_command(self, command): """ Helper method which sends the supplied command string to the commands resource. Returns the handled response to that command. """ return self.call_and_handle_commands(command)[0] def call_and_handle_commands(self, *commands): """ Helper method which sends the supplied command string to the commands resource. Returns the handled response to that command. """ response = self.call_commands(*commands) return [ self.result_handler.handle_commands_response(r[JSON_FEEDBACK]) for r in response] def call_commands(self, *commands): """ Helper method which sends the supplied command string to the commands resource. Returns the handled response to that command. """ response = self.connection.request_commands(*commands) return response[JSON_COMMANDS]