mirror of
https://github.com/Xevion/v6-place.git
synced 2025-12-06 13:16:56 -06:00
49 lines
1.2 KiB
Python
49 lines
1.2 KiB
Python
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
import asyncio
|
|
import io
|
|
import os
|
|
|
|
from PIL import Image
|
|
|
|
from place.client import PlaceClient
|
|
from place.constants import Environment
|
|
|
|
|
|
# Start a websocket
|
|
# In the initial image load, detect all changes between the target and
|
|
|
|
|
|
async def get_image(websocket):
|
|
while True:
|
|
data = await websocket.recv()
|
|
if type(data) == bytes:
|
|
return Image.open(io.BytesIO(data))
|
|
|
|
|
|
async def main():
|
|
# Grab the image we want to setup
|
|
width, height = int(os.getenv(Environment.CANVAS_HEIGHT)), int(os.getenv(Environment.CANVAS_HEIGHT))
|
|
original_image = Image.open(os.getenv(Environment.SOURCE_FILE))
|
|
original_image = original_image.resize((width, height), Image.LANCZOS)
|
|
|
|
# Start connection and get client connection protocol
|
|
client = await PlaceClient.connect(os.getenv(Environment.WEBSOCKET_ADDRESS))
|
|
client.current_target = original_image
|
|
asyncio.create_task(client.receive())
|
|
|
|
await asyncio.sleep(0.8)
|
|
async with client.lock():
|
|
client.source.save("./x.png")
|
|
client.current_target.save("./target.png")
|
|
await client.complete(1)
|
|
print('Complete.')
|
|
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|