python类encode()的实例源码

recorder.py 文件源码 项目:lightstep-tracer-python 作者: lightstep 项目源码 文件源码 阅读 33 收藏 0 点赞 0 评论 0
def _construct_report_request(self):
        """Construct a report request."""
        report = None
        with self._mutex:
            report = ttypes.ReportRequest(self._runtime, self._span_records,
                                          None)
            self._span_records = []
        for span in report.span_records:
            for log in span.log_records:
                index = span.log_records.index(log)
                if log.payload_json is not None:
                    try:
                        log.payload_json = \
                                           jsonpickle.encode(log.payload_json,
                                                             unpicklable=False,
                                                             make_refs=False,
                                                             max_depth=constants.JSON_MAX_DEPTH)
                    except:
                        log.payload_json = jsonpickle.encode(constants.JSON_FAIL)
                span.log_records[index] = log
        return report
APIHelper.py 文件源码 项目:pepipost-sdk-python 作者: pepipost 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def form_encode_parameters(form_parameters):       
        """Form encodes a dictionary of form parameters

        Args:
            form_parameters (dictionary): The given dictionary which has
            atleast one model to form encode.

        Returns:
            dict: A dictionary of form encoded properties of the model.

        """
        encoded = dict()
        for key, value in form_parameters.items():
            encoded.update(APIHelper.form_encode(value, key))

        return encoded
api_helper.py 文件源码 项目:apimatic-cli 作者: apimatic 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def form_encode_parameters(form_parameters,
                               array_serialization="indexed"):
        """Form encodes a dictionary of form parameters

        Args:
            form_parameters (dictionary): The given dictionary which has
            atleast one model to form encode.
            array_serialization (str): The format of array parameter serialization.

        Returns:
            dict: A dictionary of form encoded properties of the model.

        """
        encoded = []

        for key, value in form_parameters.items():
            encoded += APIHelper.form_encode(value, key, array_serialization)

        return encoded
web_handler.py 文件源码 项目:DecidePolitics 作者: georgeaf99 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def handle_message(customer_uuid):
    """Handles messages from a phone number with body

    @param customer_uuid The UUID of the sender

    @data message_body The body of the message
    """

    data = jsonpickle.decode(request.data.decode("utf-8"))
    message_body = data["message_body"]

    # Retrieve the message and create the customer
    customer = Customer.load_from_db(customer_uuid)
    messaging.on_message_recieve(customer, message_body)

    return jsonpickle.encode(dict(
        success=True
    ))
twillio_handler.py 文件源码 项目:DecidePolitics 作者: georgeaf99 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def handle_sms():
    """Handles Twillio SMS message"""
    customer_phone_number = request.values["From"]
    text_message_body = request.values["Body"]

    # Retrieve the customer from the DB or create a new one
    customer = Customer.get_customer_by_phone_number(customer_phone_number)
    if not customer:
        # Save the customer to the DB
        customer = Customer.create_new({
            CFields.PHONE_NUMBER: customer_phone_number,
        })
        customer.create()

        customer_lc.on_new_customer()

    messaging.on_message_recieve(customer, text_message_body)

    return jsonpickle.encode(dict(
        success=True
    ))
scheduler.py 文件源码 项目:Cayenne-Agent 作者: myDevicesIoT 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def ToJson(object):
    returnValue = "{}"
    try:

        returnValue = encode(object, unpicklable=False, make_refs=False)
    except:
        exception('ToJson Failed')
    return returnValue
#{
#   "id":"some_unique_id",
#   "type":"date/interval",
#   "unit":"minute/hour/day/week/month/year",
#   "interval":1,
#   "weekday":"monday/tuesday/...",
#   "start_date":"date and hour",
#   "end_date":"date and hour",
#   "title":"sometext",
#   "notify":"yes/no",
#   "actions":
#   [
#   ]
#} 

#AccountId = None
wifisetup.py 文件源码 项目:Cayenne-Agent 作者: myDevicesIoT 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def Search(self):
        cells = Cell.all('wlan0')
        self.endpoints = list(cells)
        serializableEndpoints = []
        for i in self.endpoints:
            wifiEndpoint = WifiEndpoint()
            wifiEndpoint.ssid = i.ssid
            wifiEndpoint.signal = i.signal
            wifiEndpoint.quality = i.quality
            wifiEndpoint.frequency = i.frequency
            wifiEndpoint.bitrates = i.bitrates
            wifiEndpoint.encrypted = i.encrypted
            wifiEndpoint.channel = i.channel
            wifiEndpoint.address = i.address
            wifiEndpoint.mode = i.mode
            serializableEndpoints.append(wifiEndpoint)
            #print (i.ssid, i.signal, i.quality, i.frequency)            
        schemes = Scheme.all()
        #print(jsonpickle.encode(schemes))
        return jsonpickle.encode(serializableEndpoints)
system_backup_handler.py 文件源码 项目:zynthian-webconf 作者: zynthian 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def on_websocket_message(self, restoreFile):
        #fileinfo = self.request.files['ZYNTHIAN_RESTORE_FILE'][0]
        #restoreFile = fileinfo['filename']
        with open(restoreFile , "rb") as f:
            validRestoreItems = get_backup_items(SYSTEM_BACKUP_ITEMS_FILE)
            validRestoreItems += get_backup_items(DATA_BACKUP_ITEMS_FILE)

            with zipfile.ZipFile(f,'r') as restoreZip:
                for member in restoreZip.namelist():
                    if self.is_valid_restore_item(validRestoreItems, member):
                        logMessage = "restored: " + member
                        restoreZip.extract(member, "/")
                        logging.debug(logMessage)
                        message = ZynthianWebSocketMessage('RestoreMessageHandler', logMessage)
                        self.websocket.write_message(jsonpickle.encode(message))
                    else:
                        logging.warn("restore of " + member + " not permitted")
                restoreZip.close()
            f.close()
        os.remove(restoreFile)
        message = ZynthianWebSocketMessage('RestoreMessageHandler', 'EOCOMMAND')
        self.websocket.write_message(jsonpickle.encode(message));
training.py 文件源码 项目:tensorlight 作者: bsautermeister 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def save(self, filepath):
        """Saves the optimizer parameters to the specifiec path as JSON.
        Parameters
        ----------
        filepath: str
            The file path.
        """
        # check and create dirs
        if not os.path.exists(os.path.dirname(filepath)):
            subdirs = os.path.dirname(filepath)
            if subdirs is not None and subdirs != '':
                os.makedirs(subdirs)

        with open(filepath, 'wb') as f:
            json = jsonpickle.encode(self)
            f.write(json)
model.py 文件源码 项目:tensorlight 作者: bsautermeister 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def save(self, filepath):
        """Saves the model parameters to the specifiec path as JSON.
        Parameters
        ----------
        filepath: str
            The file path.
        """
        # check and create dirs
        if not os.path.exists(os.path.dirname(filepath)):
            subdirs = os.path.dirname(filepath)
            if subdirs is not None and subdirs != '':
                os.makedirs(subdirs)

        with open(filepath, 'wb') as f:
            json = jsonpickle.encode(self)
            f.write(json)
deck.py 文件源码 项目:pyCardDeck 作者: iScrE4m 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def _get_exported_string(format_stripped: str, deck: Deck) -> str:
    """
    Helper function to Deck.export()

    :param format_stripped:     Desired format stripped of any spaces and lowercase
    :type format_stripped:      str
    :param deck:                instance of a Deck
    :type deck:                 :ref:`Deck`
    :return:                    YAML/JSON string of the deck
    :rtype:                     str
    :raises UnknownFormat:      when it doesn't recognize format_stripped
    """
    if format_stripped == "yaml" or format_stripped == "yml":
        exported = yaml.dump(deck)
        log.debug("Exported deck %r to a yaml string", deck)
    elif format_stripped == "json":
        exported = jsonpickle.encode(deck)
        log.debug("Exported deck %r to a json string", deck)
    else:
        log.debug("Unknown format: %s", format)
        raise UnknownFormat("Unknown format: {}".format(format))
    return exported
api_helper.py 文件源码 项目:moesifapi-python 作者: Moesif 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def form_encode_parameters(form_parameters):
        """Form encodes a dictionary of form parameters

        Args:
            form_parameters (dictionary): The given dictionary which has
            atleast one model to form encode.

        Returns:
            dict: A dictionary of form encoded properties of the model.

        """
        encoded = dict()
        for key, value in form_parameters.items():
            encoded.update(APIHelper.form_encode(value, key))

        return encoded
save.py 文件源码 项目:Imperialist-Bastards 作者: alabecki 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def save_game(save_path, players, relations, market, provinces):

    state = dict()
    jsonpickle.set_preferred_backend('simplejson')

    #save = shelve.open(file_name, flag = "n", writeback = False)
    for p, player in players.items():
        state[p] = player
    for re, rel in relations.items():
        state[re] = rel
    #state["relations"] = relations
    for p, prov in provinces.items():
        state[prov.name] = prov
    #for uc, unciv in uncivilized_minors.items():
    #   state[uc] = unciv
    state["market"] = market
    with open(save_path, 'w') as save:
        save.write(jsonpickle.encode(state, keys = True, warn = True))

    save.close()
    #save.close()

    #print("%s saved to disk \n" % (file_name))
gc_ms.py 文件源码 项目:edd 作者: JBEI 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def find_peaks_automatically_and_export (self,
      n_expected=None,
      include_headers=False) :
    err = StringIO()
    peak_times, bandwidth = self.find_consensus_peaks(
      n_expected=n_expected, err=err)
    table, errors = self.extract_peak_areas(peak_times, bandwidth, err=err)
    if include_headers :
      table.insert(0, ["Sample ID"] +
                      ["Peak %d" % (i+1) for i in range(len(peak_times)) ])
      table.insert(1, [None] + ["%.4fm" % x for x in peak_times])
    return {
      "data_type" : "gc_ms",
      "auto_peak" : True,
      "bandwidth" : bandwidth,
      "bandwidth_auto" : True, #bandwidth_auto,
      "peak_times" : peak_times,
      "sample_data" : table,
      "errors" : errors,
      "samples" : jsonpickle.encode(self.samples),
    }
gc_ms.py 文件源码 项目:edd 作者: JBEI 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def find_peaks_by_range_and_export (self, rt_ranges,
      molecule_names=None) :
    assert (molecule_names is None) or (len(molecule_names) == len(rt_ranges))
    table, errors = self.extract_peak_areas_by_range(rt_ranges)
    peak_ranges = ["%.4f - %.4fm" %(x,y) for (x,y) in rt_ranges]
    if (molecule_names is not None) :
      table.insert(0, ["Sample ID"] + list(molecule_names))
      table.insert(1, [None] + peak_ranges)
    return {
      "data_type" : "gc_ms",
      "auto_peak" : False,
      "peak_times" : peak_ranges,
      "sample_data" : table,
      "errors" : errors,
      "samples" : jsonpickle.encode(self.samples),
    }
sysinfo.py 文件源码 项目:sia-cog 作者: deepakkumar1984 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def getGPUUsage():
    try:
        pynvml.nvmlInit()
        count = pynvml.nvmlDeviceGetCount()
        if count == 0:
            return None

        result = {"driver": pynvml.nvmlSystemGetDriverVersion(),
                  "gpu_count": int(count)
                  }
        i = 0
        gpuData = []
        while i<count:
            handle = pynvml.nvmlDeviceGetHandleByIndex(i)
            mem = pynvml.nvmlDeviceGetMemoryInfo(handle)
            gpuData.append({"device_num": i, "name": pynvml.nvmlDeviceGetName(handle), "total": round(float(mem.total)/1000000000, 2), "used": round(float(mem.used)/1000000000, 2)})
            i = i+1

        result["devices"] = jsonpickle.encode(gpuData, unpicklable=False)
    except Exception as e:
        result = {"driver": "No GPU!", "gpu_count": 0, "devices": []}

    return result
objcls.py 文件源码 项目:sia-cog 作者: deepakkumar1984 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def predict(imagepath, target_x, target_y, name, model):
    if imagepath.startswith('http://') or imagepath.startswith('https://') or imagepath.startswith('ftp://'):
        response = requests.get(imagepath)
        img = Image.open(BytesIO(response.content))
        img = img.resize((target_x, target_y))
    else:
        if not os.path.exists(imagepath):
            raise Exception('Input image file does not exist')
        img = image.load_img(imagepath, target_size=(target_x, target_y))

    x = image.img_to_array(img)
    x = np.expand_dims(x, axis=0)
    x = processInputImage(name, x)
    preds = decodePrediction(name, model.predict(x))
    result = []
    for p in preds[0]:
        result.append({"synset": p[0], "text": p[1], "prediction": float("{0:.2f}".format((p[2] * 100)))})

    return json.loads(jsonpickle.encode(result, unpicklable=False))
intentapi.py 文件源码 项目:sia-cog 作者: deepakkumar1984 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def predictint():
    message = "Success"
    code = 200
    result = []
    try:
        start = datetime.utcnow()
        data = request.args.get('data')
        result = intentanalyzer.predict(data)
        result = json.loads(jsonpickle.encode(result, unpicklable=False))
        logmgr.LogPredSuccess("intent", constants.ServiceTypes.LangIntent, start)
    except Exception as e:
        code = 500
        message = str(e)
        logmgr.LogPredError("intent", constants.ServiceTypes.LangIntent, start, message)

    return jsonify({"statuscode": code, "message": message, "result": result})
nltkmgr.py 文件源码 项目:sia-cog 作者: deepakkumar1984 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def tokenize(data, language="english", filterStopWords = False, tagging = False):
    result = {}
    tags = []
    filterChars = [",", ".", "?", ";", ":", "'", "!", "@", "#", "$", "%", "&", "*", "(", ")", "+", "{", "}", "[", "]", "\\", "|"]
    sent_token = nltk.tokenize.sent_tokenize(data, language)
    word_token = nltk.tokenize.word_tokenize(data, language)
    word_token = [w for w in word_token if not w in filterChars]
    if filterStopWords is True:
        stop_words = set(stopwords.words(language))
        word_token = [w for w in word_token if not w in stop_words]

    if tagging is True:
        tags = nltk.pos_tag(word_token)

    result = {"sent_token": sent_token, "word_token": word_token, "pos_tag": tags}
    return json.loads(jsonpickle.encode(result, unpicklable=False))
live_mame_processing.py 文件源码 项目:cps2-gfx-editor 作者: goosechooser 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def main(p, folder):
    save_ = BASE_PATH + folder
    while p.poll() is None:
        result = await process_frame()
        if result:
            file_name = 'frame_' + str(result[0]) + '.txt'
            with open(save_ + file_name, 'w') as f:
                for res in result[1]:
                    f.write(jsonpickle.encode(res) + "\n")

    remaining = Q.qsize()
    print(str(remaining) + " frames left")

    for i in range(remaining):
        result = await process_frame()
        if result:
            print(str(Q.qsize()) + " left in the queue")
            file_name = 'frame_' + str(result[0]) + '.txt'
            with open(save_ + file_name, 'w') as f:
                for res in result[1]:
                    f.write(jsonpickle.encode(res) + "\n")
test_gui.py 文件源码 项目:fomod-designer 作者: GandaG 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def test_read_settings(mock_open):
    mock_open.return_value = StringIO(encode(default_settings))
    assert read_settings() == default_settings

    broken_settings = deepcopy(default_settings)
    broken_settings["General"] = "random string"  # simulate user messing with settings
    mock_open.return_value = StringIO(encode(broken_settings))
    assert read_settings() == default_settings

    mock_open.side_effect = FileNotFoundError("mock settings file not existing")
    assert read_settings() == default_settings

    mock_open.side_effect = JSONDecodeError(
        "mock settings file not being decodable - someone messed with the file",
        encode(default_settings),
        10  # just a random value
    )
    assert read_settings() == default_settings
api_helper.py 文件源码 项目:MundiAPI-PYTHON 作者: mundipagg 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def form_encode_parameters(form_parameters,
                               array_serialization="indexed"):
        """Form encodes a dictionary of form parameters

        Args:
            form_parameters (dictionary): The given dictionary which has
            atleast one model to form encode.
            array_serialization (str): The format of array parameter serialization.

        Returns:
            dict: A dictionary of form encoded properties of the model.

        """
        encoded = []

        for key, value in form_parameters.items():
            encoded += APIHelper.form_encode(value, key, array_serialization)

        return encoded
protocol.py 文件源码 项目:pyvizio 作者: vkorn 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def invoke_api(ip, command, logger, headers=None):
    if headers is None:
        headers = {}

    try:
        method = command.get_method()
        url = "https://{0}:9000{1}".format(ip, command.get_url())
        data = jsonpickle.encode(command, unpicklable=False)
        if "get" == method.lower():
            response = requests.get(url=url, headers=headers, verify=False)
        else:
            headers["Content-Type"] = "application/json"
            response = requests.request(method=method, data=str(data), url=url, headers=headers, verify=False)
        json_obj = validate_response(response)
        return command.process_response(json_obj)
    except Exception as e:
        logger.error("Failed to execute command: %s", e)
        return None
config.py 文件源码 项目:service-fabric-cli 作者: Azure 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def set_aad_cache(token, cache):
    """Set AAD token cache."""
    set_config_value('aad_token', jsonpickle.encode(token))
    set_config_value('aad_cache', jsonpickle.encode(cache))
format.py 文件源码 项目:treecat 作者: posterior 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def json_dumps(value):
    return jsonpickle.encode(value)
format.py 文件源码 项目:treecat 作者: posterior 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def fingerprint(obj):
    serialized = json_dumps(obj)
    with open('/tmp/v1.json', 'w') as f:
        print('see /tmp/v1.json')
        f.write(serialized)
    hasher = hashlib.sha1()
    try:
        hasher.update(serialized)
    except TypeError:
        hasher.update(serialized.encode('utf-8'))
    return hasher.hexdigest()
format.py 文件源码 项目:treecat 作者: posterior 项目源码 文件源码 阅读 40 收藏 0 点赞 0 评论 0
def csv_reader(filename, encoding='utf-8'):
    with io.open(filename, 'r', encoding=encoding) as f:
        if sys.version_info >= (3, 0):
            yield csv.reader(f)
        else:
            yield csv.reader(line.encode(encoding, 'ignore') for line in f)
views.py 文件源码 项目:GAMADV-XTD 作者: taers232c 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def _make_flow(request, scopes, return_url=None):
    """Creates a Web Server Flow

    Args:
        request: A Django request object.
        scopes: the request oauth2 scopes.
        return_url: The URL to return to after the flow is complete. Defaults
            to the path of the current request.

    Returns:
        An OAuth2 flow object that has been stored in the session.
    """
    # Generate a CSRF token to prevent malicious requests.
    csrf_token = hashlib.sha256(os.urandom(1024)).hexdigest()

    request.session[_CSRF_KEY] = csrf_token

    state = json.dumps({
        'csrf_token': csrf_token,
        'return_url': return_url,
    })

    flow = client.OAuth2WebServerFlow(
        client_id=django_util.oauth2_settings.client_id,
        client_secret=django_util.oauth2_settings.client_secret,
        scope=scopes,
        state=state,
        redirect_uri=request.build_absolute_uri(
            urlresolvers.reverse("google_oauth:callback")))

    flow_key = _FLOW_KEY.format(csrf_token)
    request.session[flow_key] = jsonpickle.encode(flow)
    return flow
models.py 文件源码 项目:GAMADV-XTD 作者: taers232c 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def get_prep_value(self, value):
        """Overrides ``models.Field`` method. This is used to convert
        the value from an instances of this class to bytes that can be
        inserted into the database.
        """
        if value is None:
            return None
        else:
            return encoding.smart_text(
                base64.b64encode(jsonpickle.encode(value).encode()))
APIHelper.py 文件源码 项目:pepipost-sdk-python 作者: pepipost 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def json_serialize(obj):
        """JSON Serialization of a given object.

        Args:
            obj (object): The object to serialise.

        Returns:
            str: The JSON serialized string of the object.

        """
        if obj is None:
            return None

        # Resolve any Names if it's one of our objects that needs to have this called on
        if isinstance(obj, list):
            value = list()

            for item in obj:
                try:
                    value.append(item.resolve_names())
                except (AttributeError, TypeError):
                    value.append(item)

            obj = value
        else:
            try:
                obj = obj.resolve_names()
            except (AttributeError, TypeError):
                obj = obj

        return jsonpickle.encode(obj, False)


问题


面经


文章

微信
公众号

扫码关注公众号