modm_data.dl.stmicro

13class Document:
14    """
15    Smaller helper class to store all relevant information about a remote
16    STMicro PDF document and how to download it.
17    """
18
19    def __init__(self, data):
20        # Primary information from JSON data
21        self._title = data["title"]
22        self._description = data["localizedDescriptions"]["en"]
23        self._url = data["localizedLinks"]["en"]
24        self._version = data["version"]
25        self._update = data["latestUpdate"]
26        self._type = data.get("resourceType", "Unknown")
27
28        # Derived information
29        self._short_type = self._title[:2]
30        clean_version = self._version.replace(".0", "").replace(".", "_")
31        self.name = self._title + "-v" + clean_version
32        """The full name of the document including version."""
33        self.filename = self.name + ".pdf"
34        """The leaf filename of the document."""
35        self.url = "https://www.st.com" + self._url
36        """The full URL that the file was downloaded from."""
37
38    @property
39    def data(self) -> dict:
40        """
41        A dictionary uniquely identifying this document version in a similar
42        format to the one used by the STMicro homepage. This can be used to
43        keep track of which documents have already been downloaded and which
44        have been updated upstream.
45        """
46        return {
47            "title": self._title,
48            "localizedDescriptions": {"en": self._description},
49            "localizedLinks": {"en": self._url},
50            "version": self._version,
51            "latestUpdate": self._update,
52            "resourceType": self._type,
53        }
54
55    def store_pdf(self, path: str, overwrite: bool = False) -> bool:
56        """Download the PDF file to the path, optionally overwriting it."""
57        return download_file(self.url, path, overwrite=overwrite)
58
59    def __repr__(self) -> str:
60        return f"Doc({self._title} v{self._version.replace('.0', '')})"
61
62    def __eq__(self, other) -> bool:
63        if not isinstance(other, type(self)):
64            return False
65        return self.filename == other.filename
66
67    def __hash__(self) -> int:
68        return hash(self.filename)

118def load_remote_info(base_dir: Path, use_cached: bool = False) -> list[dict]:
119    info = base_dir / _remote_info
120    if use_cached and info.exists():
121        LOGGER.debug("Loading remote info from cache")
122        docs = json.loads(info.read_text())
123    else:
124        LOGGER.info("Downloading remote info")
125        docs = []
126        for urls in _json_urls.values():
127            for url in urls:
128                docs.extend(json.loads(download_data(url))["rows"])
129    missing = [d for d in docs if "title" not in d or "version" not in d]
130    if missing:
131        LOGGER.warning(
132            "Some documents are missing title or version information and will be ignored: %s",
133            str(missing),
134        )
135    return [d for d in docs if "title" in d and "version" in d]

145def load_local_info(base_dir: Path) -> list[dict]:
146    info = base_dir / _local_info
147    if info.exists():
148        LOGGER.debug("Loading local info from cache")
149        return json.loads(info.read_text())
150    return []

138def store_remote_info(base_dir: Path, docs: list[dict]):
139    info = base_dir / _remote_info
140    info.parent.mkdir(parents=True, exist_ok=True)
141    exit(1)
142    info.write_text(json.dumps(sorted(docs, key=lambda d: (d["title"], d["version"])), indent=4, sort_keys=True))

153def store_local_info(base_dir: Path, docs: list[dict]):
154    info = base_dir / _local_info
155    info.parent.mkdir(parents=True, exist_ok=True)
156    info.write_text(json.dumps(sorted(docs, key=lambda d: (d["title"], d["version"])), indent=4, sort_keys=True))

28def download_cubemx(extraction_path: Path, with_download: bool = True, with_patch: bool = True) -> bool:
29    # First check STMUpdaterDefinitions.xml from this zip
30    update_url = "https://sw-center.st.com/packs/resource/utility/updaters.zip"
31    update_url2 = "https://www.ebuc23.com/s3/stm_test/software/utility/updaters.zip"  # noqa: F841
32    # Then Release="MX.6.2.0" maps to this: -win, -lin, -mac
33    cube_url = "https://sw-center.st.com/packs/resource/library/stm32cube_mx_v{}-lin.zip"
34    cube_url2 = "https://www.ebuc23.com/s3/stm_test/software/library/stm32cube_mx_v{}-lin.zip"  # noqa: F841
35
36    if with_download:
37        LOGGER.info("Downloading Update Info...")
38        # try:
39        #     urllib.request.urlopen(urllib.request.Request(update_url, headers=_hdr))
40        # except:
41        #     update_url = update_url2
42        #     cube_url = cube_url2
43        # LOGGER.debug(update_url)
44        # time.sleep(random.randrange(0,2))
45
46        z = zipfile.ZipFile(io.BytesIO(_dl(update_url)))
47        with io.TextIOWrapper(z.open("STMUpdaterDefinitions.xml"), encoding="utf-8") as defs:
48            version = re.search(r'Release="MX\.(.*?)"', defs.read())
49            version = version.group(1).replace(".", "")
50
51        shutil.rmtree(extraction_path / "mcu", ignore_errors=True)
52        shutil.rmtree(extraction_path / "plugins", ignore_errors=True)
53        extraction_path.mkdir(exist_ok=True, parents=True)
54
55        LOGGER.info("Downloading Database...")
56        LOGGER.debug(cube_url.format(version))
57        time.sleep(random.randrange(1, 6))
58
59        z = zipfile.ZipFile(io.BytesIO(_dl(cube_url.format(version))))
60        LOGGER.info("Extracting Database...")
61        for file in z.namelist():
62            if any(file.startswith(prefix) for prefix in ("MX/db/mcu", "MX/db/plugins")):
63                z.extract(file, extraction_path)
64
65        LOGGER.info("Moving Database...")
66        shutil.move(extraction_path / "MX/db/mcu", extraction_path / "mcu")
67        shutil.move(extraction_path / "MX/db/plugins", extraction_path / "plugins")
68        shutil.rmtree(extraction_path / "MX", ignore_errors=True)
69
70        LOGGER.info("Normalizing file endings...")
71        for file in Path(extraction_path).glob("**/*"):
72            if str(file).endswith(".xml"):
73                with file.open("r", newline=None, encoding="utf-8", errors="replace") as rfile:
74                    content = [line.rstrip() + "\n" for line in rfile.readlines()]
75                with file.open("w", encoding="utf-8") as wfile:
76                    wfile.writelines(content)
77
78    if with_patch:
79        LOGGER.info("Patching Database...")
80        from . import data
81
82        return pkg_apply_patch(data, "cubemx.patch", extraction_path)
83
84    return True

19def download_cubeprog(extraction_path: Path, with_download: bool = True) -> bool:
20    # com.st.stm32cube.ide.mcu.productdb.debug contains all the SVD files
21    # com.st.stm32cube.ide.mcu.externaltools.cubeprogrammer.linux64 contains the CubeProgrammer database
22    # com.st.stm32cube.common.mx contains the CubeMX database
23    base_url = "http://sw-center.st.com/stm32cubeide/updatesite1/"
24    cubeprog_id = "com.st.stm32cube.ide.mcu.externaltools.cubeprogrammer.linux64"
25
26    if with_download:
27        LOGGER.info("Downloading Update Info...")
28        with io.TextIOWrapper(
29            io.BytesIO(_dl(f"{base_url}/compositeContent.xml")), encoding="utf-8"
30        ) as compositeContent:
31            update_version = re.findall(r"child location='(\d+\.\d+\.\d+)'", compositeContent.read())
32            update_version = sorted(update_version, key=lambda v: tuple(map(int, v.split("."))), reverse=True)[0]
33
34        artifacts = lzma.LZMADecompressor().decompress(_dl(f"{base_url}/{update_version}/artifacts.xml.xz"))
35        version = re.search(f"id='{cubeprog_id}' version='(.*?)'", artifacts.decode("utf-8")).group(1)
36
37        download_url = f"{base_url}/{update_version}/plugins/{cubeprog_id}_{version}.jar"
38        shutil.rmtree(extraction_path / "Data_Base", ignore_errors=True)
39        extraction_path.mkdir(exist_ok=True, parents=True)
40
41        LOGGER.info("Downloading Database...")
42        LOGGER.debug(download_url)
43        time.sleep(random.randrange(1, 6))
44
45        z = zipfile.ZipFile(io.BytesIO(_dl(download_url)))
46        LOGGER.info("Extracting Database...")
47        for file in z.namelist():
48            if file.startswith("tools/Data_Base"):
49                z.extract(file, extraction_path)
50
51        LOGGER.info("Moving Database...")
52        shutil.move(extraction_path / "tools/Data_Base", extraction_path / "Data_Base")
53        shutil.rmtree(extraction_path / "tools", ignore_errors=True)
54
55        LOGGER.info("Normalizing file endings...")
56        for file in Path(extraction_path).glob("**/*"):
57            if str(file).endswith(".xml"):
58                with file.open("r", newline=None, encoding="utf-8", errors="replace") as rfile:
59                    content = [line.rstrip() + "\n" for line in rfile.readlines()]
60                with file.open("w", encoding="utf-8") as wfile:
61                    wfile.writelines(content)
62
63    return True