infeeeee/dyn2py
1
0
Fork 0
mirror of https://github.com/infeeeee/dyn2py synced 2025-12-16 22:16:18 +01:00
Code Issues Packages Projects Releases Activity

New tests, small changes on a lot of things

Browse Source
This commit is contained in:
Péter Gyetvai
2023-03-16 02:31:28 +01:00
parent d6fa3d25db
commit 9a7eb92eb4
10 changed files with 350 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
dyn2py/__init__.py
View File

@@ -8,6 +8,7 @@ import pathlib
from importlib_metadata import metadata
import textwrap
import logging
import inspect
from dyn2py.files import *
from dyn2py.options import *
@@ -32,100 +33,129 @@ def __dir__():
return __all__
def run(options: Options | None = None) -> None:
def __command_line() -> None:
"""Private method for running as a console script"""
parser = argparse.ArgumentParser(
prog=METADATA["Name"],
description=METADATA["Summary"],
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=textwrap.dedent("""\
The script by default overwrites older files with newer files.
Do not move the source Dynamo graphs, or update won't work with them later.
Multiple sources are supported, separate them by spaces.
""")
)
parser.add_argument("-v", "--version",
action="version",
version=f'{METADATA["Name"]} {METADATA["Version"]}'
)
parser.add_argument("-l", "--loglevel",
metavar="LOGLEVEL",
choices=LOGLEVELS,
default=DEFAULT_LOGLEVEL,
help=f"set log level, possible options: {', '.join(LOGLEVELS)} ")
parser.add_argument("-n", "--dry-run",
help="do not modify files, only show log",
action="store_true"
)
parser.add_argument("-F", "--force",
help="overwrite even if the files are older",
action="store_true")
parser.add_argument("-b", "--backup",
help="create a backup for updated files",
action="store_true")
parser.add_argument("-f", "--filter",
choices=FILTERS,
help="only check python or Dynamo graphs, skip the others, useful for folders"
)
dynamo_options = parser.add_argument_group(
title="dynamo options, only for processing Dynamo graphs")
dynamo_options.add_argument("-u", "--update",
help="update Dynamo graph from python scripts in the same folder",
action="store_true")
dynamo_options.add_argument("-p", "--python-folder",
metavar="path/to/folder",
help="extract python scripts to this folder, read python scripts from here with --update",
type=pathlib.Path)
parser.add_argument("source",
type=pathlib.Path,
help="path to a Dynamo graph, a python script or a folder containing them",
nargs="*"
)
options = parser.parse_args(namespace=Options())
run(options)
def run(options: Options) -> None:
"""Run an extraction as from the command line
Args:
options (Options): Options as from the command line.
Raises:
TypeError: options is not an Options object
FileNotFoundError: If the source file does not exist
"""
if not options:
if not isinstance(options, Options):
raise TypeError("Options have to be a dyn2py.Options() object!")
parser = argparse.ArgumentParser(
prog=METADATA["Name"],
description=METADATA["Summary"],
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=textwrap.dedent("""\
The script by default overwrites older files with newer files.
Do not move the source Dynamo graphs, or update won't work with them later.
""")
)
parser.add_argument("-v", "--version",
action="version",
version=f'{METADATA["Name"]} {METADATA["Version"]}'
)
parser.add_argument("-l", "--loglevel",
metavar="LOGLEVEL",
choices=LOGLEVELS,
default=DEFAULT_LOGLEVEL,
help=f"set log level, possible options: {', '.join(LOGLEVELS)} ")
parser.add_argument("-n", "--dry-run",
help="do not modify files, only show log",
action="store_true"
)
parser.add_argument("-F", "--force",
help="overwrite even if the files are older",
action="store_true")
parser.add_argument("-b", "--backup",
help="create a backup for updated files",
action="store_true")
parser.add_argument("-f", "--filter",
choices=FILTERS,
help="only check python or Dynamo graphs, skip the others, useful for folders"
)
dynamo_options = parser.add_argument_group(
title="dynamo options, only for processing Dynamo graphs")
dynamo_options.add_argument("-u", "--update",
help="update Dynamo graph from python scripts in the same folder",
action="store_true")
dynamo_options.add_argument("-p", "--python-folder",
metavar="path/to/folder",
help="extract python scripts to this folder, read python scripts from here with --update",
type=pathlib.Path)
parser.add_argument("source",
type=pathlib.Path,
help="path to a Dynamo graph, a python script or a folder containing them",
action="append"
)
options = parser.parse_args(namespace=Options())
from_command_line = bool(inspect.stack()[1].function == "__command_line")
# Set up logging:
logging.basicConfig(format='%(levelname)s: %(message)s',
level=options.loglevel)
logging.debug(options)
logging.debug(f"Parsed arguments: {vars(options)}")
logging.debug(f"Run options: {vars(options)}")
# Set up sources:
files = []
source_files = []
for source in options.source:
if not source.exists():
raise FileNotFoundError(f"Source file does not exist!")
if from_command_line:
# log only if it was called from command line:
logging.error(f"File does not exist: {source}")
exit(1)
else:
raise FileNotFoundError(f"Source file does not exist!")
# Get files from folder:
elif source.is_dir():
logging.debug(f"Source is a folder")
for f in source.iterdir():
files.append(File(f))
source_files.append(f)
# It's a single file:
else:
files.append(File(source))
source_files.append(source)
# Create file objects
files = []
for f in source_files:
try:
files.append(File(f))
except DynamoFileException as e:
# It's a dynamo1 file
logging.warning(e)
continue
except PythonNodeNotFoundException as e:
# No python node in this file
logging.warning(e)
continue
# Dynamo files come first, sort sources:
files.sort(key=lambda f: f.extension)
@@ -149,21 +179,21 @@ def run(options: Options | None = None) -> None:
files = list(python_files)
if not files and from_command_line:
logging.error("No files to process! See previous warnings!")
exit(1)
# Cycle through files:
for f in files:
if f.is_dynamo_file():
logging.debug("Source is a Dynamo file")
try:
f.extract_python(options)
except DynamoFileException as e:
logging.error(f"{e} Skipping")
f.extract_python(options)
elif f.is_python_file():
logging.debug("Source is a Python file")
f.update_dynamo(options)
# Dynamo files are written only at the end, so they don't get updated too frequently
for f in DynamoFile.open_files:
# Write files at the end:
for f in DynamoFile.open_files | PythonFile.open_files:
f.write(options)

3
dyn2py/exceptions.py
View File

@@ -1,10 +1,13 @@
class DynamoFileException(Exception):
"""Something wrong in this DynamoFile"""
pass
class PythonNodeNotFoundException(Exception):
"""PythonNode not found"""
pass
class PythonNodeException(Exception):
pass

80
dyn2py/files.py
View File

@@ -22,11 +22,12 @@ class File():
"""Base class for managing files"""
def __init__(self, filepath: pathlib.Path | str, read_from_disk: bool = True) -> None:
"""Generate a file object. If the path is correct it will become a DynamoFile or PythonFile object
"""Generate a file object. If the path is correct it will become a DynamoFile or PythonFile object.
Calls DynamoFile.read_file() and PythonFile.read_file()
Args:
filepath (pathlib.Path | str): Path to the python file or Dynamo graph
read_from_disk (bool, optional): Read the file from disk. Set to false to get only metadata. Defaults to True.
read_from_disk (bool, optional): Read the file from disk. False to get only metadata. Defaults to True.
"""
self.filepath: pathlib.Path
@@ -69,7 +70,7 @@ class File():
if self.exists and read_from_disk:
self.read_file()
if self.exists and read_from_disk:
if self.exists:
logging.debug(f"File exists: {self.filepath}")
self.mtime = self.filepath.stat().st_mtime
self.mtimeiso = datetime.fromtimestamp(self.mtime).isoformat()
@@ -78,11 +79,11 @@ class File():
"""Should be implemented in subclasses"""
pass
def is_newer(self, other_file: "File") -> bool:
def is_newer(self, other_file: File) -> bool:
"""Check if this file is newer than the other file
Args:
other_file(File): The other file
other_file (File): The other file
Returns:
bool: True if this file is newer or the other doesn't exist
@@ -110,18 +111,21 @@ class File():
"""
return bool(self.extension == ".py")
def write(self, options: Options = Options()) -> None:
def write(self, options: Options | None = None) -> None:
"""Prepare writing file to the disk:
create backup, process dry-run, call filetype specific write_file() methods
Should be called on subclasses!
Args:
options(Options, optional): Run options. Defaults to Options().
options (Options | None, optional): Run options. Defaults to None.
Raises:
TypeError: If called on a File object
"""
if not options:
options = Options()
# This should only work on subclasses:
if type(self).__name__ == "File":
raise TypeError("This method shouldn't be called on File objects!")
@@ -144,11 +148,16 @@ class File():
f"Should write file, but it's a dry-run: {self.filepath}")
else:
logging.info(f"Writing file: {self.filepath}")
self.write_file()
self._write_file()
def write_file(self):
"""Should be implemented in subclasses"""
pass
def _write_file(self):
"""Should be implemented in subclasses
Raises:
NotImplementedError: If called on a File object
"""
raise NotImplementedError(
"Should be called only on DynamoFile and PythonFile objects!")
class DynamoFile(File):
@@ -166,14 +175,21 @@ class DynamoFile(File):
open_files: set[DynamoFile] = set()
"""A set of open Dynamo files, before saving. Self added by read()"""
def extract_python(self, options: Options = Options()) -> None:
"""Extract and write python files
def extract_python(self, options: Options | None = None) -> list[PythonFile]:
"""Extract python files from Dynamo graphs, add them to open_files
Args:
options(Options, optional): Run options. Defaults to Options().
options (Options | None, optional): Run options. Defaults to None.
Returns:
list[PythonFile]: List of PythonFile objects extracted from this DynamoFile
"""
if not options:
options = Options()
logging.info(f"Extracting from file: {self.filepath}")
python_files = []
# Go through nodes in the file:
for python_node in self.python_nodes:
@@ -190,11 +206,14 @@ class DynamoFile(File):
)
if python_file.is_newer(self) and not options.force:
logging.info(
PythonFile.open_files.remove(python_file)
logging.warning(
f"Existing file is newer, skipping: {python_file.filepath}")
continue
python_file.write(options)
python_files.append(python_file)
return python_files
def read_file(self, reread: bool = False) -> None:
"""Read Dynamo graph to parameters. Automatically called by __init__()
@@ -251,11 +270,11 @@ class DynamoFile(File):
dynamo_file=self)
self.python_nodes.add(python_node)
def get_python_node_by_id(self, node_id: str) -> "PythonNode":
def get_python_node_by_id(self, node_id: str) -> PythonNode:
"""Get a PythonNode object from this Dynamo graph, by its id
Args:
node_id(str): The id of the python node as string
node_id (str): The id of the python node as string
Returns:
PythonNode: The PythonNode with the given id
@@ -278,7 +297,7 @@ class DynamoFile(File):
"""Update the code of a PythonNode in this file
Args:
python_node(PythonNode): The new node
python_node (PythonNode): The new node
Raises:
PythonNodeNotFoundException: Existing node not found
@@ -303,20 +322,22 @@ class DynamoFile(File):
self.modified = True
def write_file(self) -> None:
def _write_file(self) -> None:
"""Write this file to the disk. Should be called only from File.write()"""
with open(self.filepath, "w", encoding="utf-8", newline="") as output_file:
json.dump(self.full_dict, output_file, indent=2, use_decimal=True)
def get_related_python_files(self, options: Options = Options()) -> list["PythonFile"]:
def get_related_python_files(self, options: Options | None = None) -> list[PythonFile]:
"""Get python files exported from this Dynamo file
Args:
options(Options, optional): Run options. Defaults to Options().
options (Options | None, optional): Run options. Defaults to None.
Returns:
list[PythonFile]: A list of PythonFile objects
"""
if not options:
options = Options()
# Find the folder of the python files
if options.python_folder:
@@ -333,11 +354,11 @@ class DynamoFile(File):
return related_python_files
@staticmethod
def get_open_file_by_uuid(uuid: str) -> "DynamoFile | None":
def get_open_file_by_uuid(uuid: str) -> DynamoFile | None:
"""Get an open Dynamo graph by its uuid
Args:
uuid(str): Uuid of the file
uuid (str): Uuid of the file
Returns:
DynamoFile: The file. None if not found
"""
@@ -357,7 +378,7 @@ class PythonFile(File):
text: str
"""Full contents of the file."""
open_files: set["PythonFile"] = set()
open_files: set[PythonFile] = set()
"""A set of open Python files."""
def __init__(self,
@@ -475,13 +496,16 @@ class PythonFile(File):
logging.debug(f"Header data from python file: {self.header_data}")
# logging.debug(f"Code from python file: {self.code}")
def update_dynamo(self, options: Options = Options()) -> None:
def update_dynamo(self, options: Options | None = None) -> None:
"""Update a the source Dynamo graph from this python script
Args:
options (Options, optional): Run options. Defaults to Options().
options (Options | None, optional): Run options. Defaults to None.
"""
if not options:
options = Options()
dynamo_file = self.get_source_dynamo_file()
new_python_node = PythonNode(python_file=self)
@@ -527,7 +551,7 @@ class PythonFile(File):
return dynamo_file
def write_file(self) -> None:
def _write_file(self) -> None:
"""Write this file to the disk. Should be called only from File.write()"""
with open(self.filepath, "w", encoding="utf-8", newline="") as output_file:
output_file.write(self.text)