Python

基于Lively wallpaper的壁纸

admin 9532 次阅读 约 2 分钟
python websocket

引言

wallpaper engine 应该都不陌生，但是资源消耗的也挺大的，相比Lively wallpaper，显得Lively wallpaper小巧，轻量，开发起来只需要了解前端知识，JavaScript，我要是记得没错的话，这个程序是咱阿三哥做的，其实也是挺优秀的。

简单的介绍

实际上这个程序只是把浏览器嵌套到了explorer.exe的窗体中，也就是子窗口，之前也尝试做过一个类似的小工具，只是没有这个强大，它可以播放视频，也可以展示HTML页面。Snipaste_2022-09-26_17-03-28.png

动手试一试

这个是基于Yoichi Kobayashi -- The Hill做的修改，感觉下面的动画很有意思。

image.png

Cpu、内存、网速等相关内容呢，js获取起来应该比较困难，js调用wmi应该也是个办法，没有去尝试，只能搞一个后端应用来获取这些东西，写入内存的话，前端和浏览器实现数据共享难度也挺大的，写入硬盘读文件或者数据库，redis，虽然也是可以，但是只要加上了IO，就感觉比较臃肿，况且只想做个简单的页面，就不考虑使用node了。所以说到这儿，数据怎么传输呢？WebSocket是一个不错的选择。

架构图如下：
image.png
之所以选择Python，是因为这是一个作为服务的出现，java的话，jvm对于资源上的消耗也是考虑在内的，所以使用python。

python代码

import asyncio
import time
import datetime
import websockets
import psutil
import json
import threading
import requests
import bs4
import logging
import os

logger = logging.getLogger("")
logFolder = os.path.join(os.getcwd(), 'logs')
if not os.path.exists(logFolder):
    os.makedirs(logFolder)
timestamp = time.strftime("%Y-%m-%d", time.localtime())
logfilename = '%s.txt' % (timestamp)
logFile = os.path.join(os.getcwd(), 'logs', logfilename)
logging.basicConfig(filename=logFile,
                    level=logging.DEBUG,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

buffer = {"code": 200,
          "message": "Success",
          "author": "ZunMX",
          "recv_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
          "result": {
              'weather': {},
              'sysinfo': {}
          }
          }


def get_network(flush=1):
    while True:
        try:
            global buffer
            sent_before = psutil.net_io_counters().bytes_sent  # 已发送的流量
            recv_before = psutil.net_io_counters().bytes_recv  # 已接收的流量
            time.sleep(1)
            sent_now = psutil.net_io_counters().bytes_sent
            recv_now = psutil.net_io_counters().bytes_recv
            sent = (sent_now - sent_before) / 1024  # 算出1秒后的差值
            recv = (recv_now - recv_before) / 1024
            buffer['result']['sysinfo']['network_upload'] = "{0}KB/s".format("%.2f" % sent)
            buffer['result']['sysinfo']['network_download'] = "{0}KB/s".format("%.2f" % recv)
            time.sleep(flush)
        except Exception as e:
            print(e)


def get_ram(flush=1):
    while True:
        try:
            global buffer
            phymem = psutil.virtual_memory()
            buffer['result']['sysinfo']['ram_percent'] = str(phymem.percent) + "%"
            buffer['result']['sysinfo']['ram_used'] = str(int(phymem.used / 1024 / 1024)) + "M"
            buffer['result']['sysinfo']['ram_total'] = str(int(phymem.total / 1024 / 1024)) + "M"
            time.sleep(flush)
        except Exception as e:
            print(e)


def get_cpu(flush=1):
    while True:
        try:
            global buffer
            buffer['result']['sysinfo']['cpu_percent'] = str(psutil.cpu_percent(1)) + "%"
            time.sleep(flush)
        except Exception as e:
            print(e)


def get_battery(flush=1):
    while True:
        try:
            global buffer
            battery = psutil.sensors_battery()
            buffer['result']['sysinfo']['battery_percent'] = str(battery.percent) + "%"
            time.sleep(flush)
        except Exception as e:
            print(e)


def get_weather(code='54499', flush=3600):
    while True:
        try:
            global buffer
            url = f'https://weather.cma.cn/web/weather/{code}.html'
            response = requests.get(url)
            url2 = f'https://weather.cma.cn/api/now/{code}'
            response2 = requests.get(url2)
            response.encoding = response.apparent_encoding
            response2.encoding = response2.apparent_encoding
            soup = bs4.BeautifulSoup(response.text, 'html.parser')
            text = soup.select("#dayList")[0].contents[1].text
            text = text.replace("\n\n", "").split("\n")
            buffer['result']['weather']['now'] = {
                'location': response2.json()["data"]["location"]["path"],
                "temperature": response2.json()["data"]["now"]['temperature'],
                "pressure": response2.json()["data"]["now"]["pressure"],
                "humidity": response2.json()["data"]["now"]["humidity"],
                "updateTime": response2.json()["data"]["lastUpdate"]
            }
            buffer['result']['weather']['week'] = str(text[0]).rstrip().replace(" ", "")
            buffer['result']['weather']['date'] = str(text[1]).rstrip().replace(" ", "")
            buffer['result']['weather']['weather'] = str(text[3]).rstrip().replace(" ", "")
            buffer['result']['weather']['weather2'] = str(text[9]).rstrip().replace(" ", "")
            buffer['result']['weather']['high'] = str(text[6]).rstrip().replace(" ", "")
            buffer['result']['weather']['low'] = str(text[7]).rstrip().replace(" ", "")
            buffer['result']['weather']['wind_direct'] = str(text[4]).rstrip().replace(" ", "")
            buffer['result']['weather']['wind_level'] = str(text[5]).rstrip().replace(" ", "")
            time.sleep(flush)
        except Exception as e:
            print(e)


def getinfo():
    threading.Thread(target=get_weather, args=(), ).start()
    threading.Thread(target=get_ram, args=(), ).start()
    threading.Thread(target=get_cpu, args=(), ).start()
    threading.Thread(target=get_network, args=(), ).start()
    threading.Thread(target=get_battery, args=(), ).start()


async def echo(websocket, path):
    async for message in websocket:
        global buffer
        print("got a message:{}".format(message))
        result = {"code": 500, "message": "Message Undefined"}
        if message == 'get_info':
            buffer['recv_date'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            buffer['result']['sysinfo']['alive_time'] = time.strftime('%d天%H时%M分%S秒', time.localtime(
                int(time.time() - psutil.boot_time())))
            result = buffer
        await websocket.send(json.dumps(result, ensure_ascii=False))


async def main():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()


threading.Thread(target=getinfo, args=(), ).start()
asyncio.get_event_loop().run_until_complete(websockets.serve(echo, "localhost", 8765))
asyncio.get_event_loop().run_forever()
# asyncio.run(main())

目前，python代码没有做异常处理，所以一旦其中发生异常，可能那个线程就断了，断了之后只能重启这个Python，否则那个数据是无法更新的，甚至这个程序就直接GG了。

前端代码

在这里，我只写WebSocket部分了。

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Document</title>
</head>

<body>
    <p id="message">

    </p>
</body>


<script>
    let websocket = null;
    function init() {
        websocket = new WebSocket("ws://127.0.0.1:8765");
        websocket.onerror = function (e) {
            init();
        }
        websocket.onmessage = function (event) {
            console.log(JSON.parse(event.data));
            document.getElementById("message").innerHTML = event.data;
        }
    }

    setInterval(function () {
        try {
            if (websocket.readyState == WebSocket.OPEN) {
                websocket.send("get_info")
            } else if (websocket.readyState == WebSocket.CLOSED) {
                init();
            }
        } catch (e) {
            console.log(e)
            init();
        }
    }, 1000)
</script>

</html>

相关文章

同分类推荐
中公教育CTS视频下载 2021 年 04 月 24 日 · 11186 次阅读 GodEyes - 网络空间资产搜索引擎开源项目 2022 年 01 月 07 日 · 11075 次阅读 企业微信-摸鱼必备，避免离开状态 2022 年 06 月 05 日 · 8934 次阅读

评论 0

还没有评论，来说点什么吧。

提交后需要审核才会公开。