modm_data.dl.stmicro

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

138def load_remote_info(base_dir: Path, use_cached: bool = False) -> list[dict]:
139    info = base_dir / _remote_info
140    if use_cached and info.exists():
141        LOGGER.debug("Loading remote info from cache")
142        docs = json.loads(info.read_text())
143    else:
144        LOGGER.info("Downloading remote info")
145        docs = []
146        for urls in _json_urls.values():
147            for url in urls:
148                docs.extend(_download_json(url)["rows"])
149    missing = [d for d in docs if "title" not in d or "version" not in d]
150    if missing:
151        LOGGER.warning(
152            "Some documents are missing title or version information and will be ignored: %s",
153            str(missing),
154        )
155    return [d for d in docs if "title" in d and "version" in d]

164def load_local_info(base_dir: Path) -> list[dict]:
165    info = base_dir / _local_info
166    if info.exists():
167        LOGGER.debug("Loading local info from cache")
168        return json.loads(info.read_text())
169    return []

158def store_remote_info(base_dir: Path, docs: list[dict]):
159    info = base_dir / _remote_info
160    info.parent.mkdir(parents=True, exist_ok=True)
161    info.write_text(json.dumps(sorted(docs, key=lambda d: (d["title"], d["version"])), indent=4, sort_keys=True))

172def store_local_info(base_dir: Path, docs: list[dict]):
173    info = base_dir / _local_info
174    info.parent.mkdir(parents=True, exist_ok=True)
175    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